wow-token-compactor.py (13546B)
1 import time 2 from decimal import Decimal 3 from functools import cache 4 from typing import Tuple, List, Dict 5 6 import boto3 7 import json 8 import os 9 import datetime 10 11 from boto3.dynamodb.conditions import Key 12 13 dynamo_region_map = { 14 'us-west-1': 'us-west-1', 15 'us-west-2': 'us-west-2', 16 'us-east-1': 'us-east-1', 17 'us-east-2': 'us-east-2', 18 'ap-south-1': 'eu-north-1', 19 'ap-northeast-3': 'ap-northeast-1', 20 'ap-northeast-2': 'ap-northeast-1', 21 'ap-southeast-1': 'ap-southeast-1', 22 'ap-southeast-2': 'ap-southeast-2', 23 'ap-northeast-1': 'ap-northeast-1', 24 'ca-central-1': 'us-east-1', 25 'eu-central-1': 'eu-north-1', 26 'eu-west-1': 'eu-west-1', 27 'eu-west-2': 'eu-west-1', 28 'eu-west-3': 'eu-west-3', 29 'eu-north-1': 'eu-north-1', 30 'sa-east-1': 'sa-east-1', 31 'eu-south-1': 'eu-north-1' 32 } # This is a rough first pass at an intelligent region selector based on what is replicated 33 if os.environ['AWS_REGION'] in dynamo_region_map: 34 local_dynamo_region = dynamo_region_map[os.environ['AWS_REGION']] 35 else: 36 local_dynamo_region = 'eu-central-1' 37 38 DYNAMO_CLIENT = boto3.resource('dynamodb', region_name=local_dynamo_region) 39 40 41 42 class PriceHistory: 43 region: str 44 flavor: str 45 _prices: Dict[str, List[Tuple[int, int]]] 46 47 def __init__(self, region: str, flavor: str): 48 self.region = region 49 self.flavor = flavor 50 self._prices = dict() 51 52 def _retrieve_compacted_prices_from_dynamo(self, year, month) -> None: 53 table = DYNAMO_CLIENT.Table('wow-token-compacted') 54 pk = f'{self.region}-{self.flavor}-{year}-{month}' 55 data = [] 56 response = table.query( 57 KeyConditionExpression=Key('region-flavor-timestamp').eq(pk) 58 ) 59 if response['Items']: 60 self._prices[f'{year}-{month}'] = [] 61 for timestamp, price in response['Items'][0]['data'].items(): 62 data.append((int(timestamp), int(price))) 63 self._prices[f'{year}-{month}'] = sorted(data, key=lambda x: x[0]) 64 65 66 def _retrieve_time_bin(self, start_time: datetime.datetime, end_time: datetime.datetime) -> List[Tuple[int, int]]: 67 scan_data = self.get_month_prices(start_time.month, start_time.year) 68 if end_time.year != start_time.year or end_time.month != start_time.month: 69 scan_data += self.get_month_prices(end_time.month, end_time.year) + scan_data 70 71 high_tuple = (0,0) 72 low_tuple = (0,0) 73 for item in scan_data: 74 if start_time.timestamp() <= item[0] < end_time.timestamp(): 75 if item[1] > high_tuple[1]: 76 high_tuple = item 77 if item[1] < low_tuple[1] or low_tuple[0] == 0: 78 low_tuple = item 79 if high_tuple[0] == 0 or low_tuple[0] == 0: 80 return [] 81 else: 82 if high_tuple[0] == low_tuple[0]: 83 return [high_tuple] 84 elif low_tuple[0] > high_tuple[0]: 85 return [high_tuple, low_tuple] 86 else: 87 return [low_tuple, high_tuple] 88 89 90 def request_time_to_datetime_pair(self, time_str: str) -> Tuple[datetime.datetime, datetime.datetime]: 91 end_time = datetime.datetime.now(datetime.timezone.utc) 92 if time_str == 'all': 93 if self.flavor == 'retail': 94 start_time = datetime.datetime.fromisoformat('2020-11-15 00:00:01.000000000+00:00') 95 else: 96 start_time = datetime.datetime.fromisoformat('2023-05-23 00:00:01.000000000+00:00') 97 elif time_str[-1] == 'd': 98 days = int(time_str[:-1]) 99 start_time = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=days) 100 elif time_str[-1] == 'm': 101 months = int(time_str[:-1]) 102 start_time = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=int(30.437*months)) 103 elif time_str[-1] == 'y': 104 years = int(time_str[:-1]) 105 start_time = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=int(365.25*years)) 106 else: 107 raise ValueError 108 return start_time, end_time 109 110 111 def binned_time(self, start_time: datetime.datetime, end_time: datetime.datetime) -> List[Tuple[int, int]]: 112 time_delta = end_time - start_time 113 hours = time_delta.days * 24 + time_delta.seconds // 3600 114 if hours > 8800: # Above a year 115 bin_size = 12 116 elif hours > 4400: # Above 6 months 117 bin_size = 6 118 elif hours > 2112: # 3 months 119 bin_size = 2 120 else: 121 bin_size = 1 122 123 _bin_start = start_time 124 _bin_end = _bin_start + datetime.timedelta(hours=bin_size, seconds=-1) 125 data = [] 126 127 while _bin_start < end_time: 128 data += self._retrieve_time_bin(_bin_start, _bin_end) 129 _bin_start = _bin_end + datetime.timedelta(hours=1) 130 _bin_end = _bin_start + datetime.timedelta(hours=bin_size, seconds=-1) 131 132 return data 133 134 135 def get_month_prices(self, month: int|str, year: int|str) -> List[Tuple[int, int]]: 136 if isinstance(month, int): 137 month = str(month) 138 139 if isinstance(year, int): 140 year = str(year) 141 142 if f'{year}-{month}' not in self._prices: 143 self._retrieve_compacted_prices_from_dynamo(year, month) 144 return self._prices[f'{year}-{month}'] 145 146 147 def retrieve_binned_prices(self, time_str: str) -> List[Tuple[int, int]]: 148 table = DYNAMO_CLIENT.Table('wow-token-compacted') 149 pk = f'{self.region}-{self.flavor}-{time_str}' 150 response = table.get_item( 151 Key={ 152 'region-flavor-timestamp': pk 153 } 154 ) 155 if 'Item' not in response: 156 return [] 157 data = [] 158 for _time, _price in response['Item']['data'].items(): 159 data.append((int(_time), int(_price))) 160 return sorted(data) 161 162 163 def write_binned_if_updated(self, time_str: str) -> bool: 164 current_binned_prices = self.retrieve_binned_prices(time_str) 165 start, end = self.request_time_to_datetime_pair(time_str) 166 binned_data = sorted(self.binned_time(start, end)) 167 if all(item in current_binned_prices for item in binned_data) and all(item in binned_data for item in current_binned_prices): 168 print(f"{time_str} No update needed") 169 return False 170 else: 171 self.write_binned_prices(time_str, binned_data) 172 return True 173 174 175 def write_binned_prices(self, time_str: str, binned_data: List[Tuple[int, int]] = None) -> None: 176 print(f'Writing compacted bin data for {self.region}-{self.flavor}-{time_str}') 177 table = DYNAMO_CLIENT.Table('wow-token-compacted') 178 pk = f'{self.region}-{self.flavor}-{time_str}' 179 dynamo_data: Dict[str, Dict[str, str]] = {} 180 for item in binned_data: 181 dynamo_data[str(item[0])] = item[1] 182 183 response = table.put_item( 184 Item={ 185 'region-flavor-timestamp': pk, 186 'data': dynamo_data 187 } 188 ) 189 190 191 @cache 192 def retrieve_prices_from_s3(region: str, flavor: str) -> List[Tuple[datetime.datetime, int]]: 193 s3 = boto3.client('s3') 194 key = f"{flavor}-{region}-price-history.json" 195 obj = s3.get_object( 196 Bucket=os.environ['S3_BUCKET'], 197 Key=key 198 ) 199 raw_data = json.loads(obj['Body'].read()) 200 data = [] 201 for datum in raw_data: 202 date = datetime.datetime.fromisoformat(f"{datum[2]['ScalarValue']}+00:00") 203 price = int(int(datum[3]['ScalarValue']) / 10000) 204 data.append((date, price)) 205 return data 206 207 208 def retrieve_recent_prices(region: str, flavor: str) -> List[Tuple[datetime.datetime, int]]: 209 start_time = (datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=95)) 210 if flavor == 'retail': 211 table = DYNAMO_CLIENT.Table('wow-token-price-recent') 212 else: 213 table = DYNAMO_CLIENT.Table('wow-token-classic-price-recent') 214 response = table.query( 215 KeyConditionExpression=( 216 Key('region').eq(region) & 217 Key('timestamp').gte(int(start_time.timestamp())))) 218 data = [] 219 last_price = 0 220 for item in response['Items']: 221 price = int(int(item['price']) / 10000) 222 if last_price != price: 223 last_price = price 224 item_time = datetime.datetime.fromtimestamp(int(item['timestamp']), tz=datetime.timezone.utc) 225 data.append((item_time, price)) 226 return data 227 228 229 def write_compacted_month(region: str, flavor: str, year: int, month: int, data: List[Tuple[datetime.datetime, int]])-> None: 230 table = DYNAMO_CLIENT.Table('wow-token-compacted') 231 pk = f'{region}-{flavor}-{year}-{month}' 232 dynamo_data: Dict[str, Dict[str, str]] = {} 233 for item in data: 234 dynamo_data[str(int(item[0].timestamp()))] = item[1] 235 236 response = table.put_item( 237 Item={ 238 'region-flavor-timestamp': pk, 239 'data': dynamo_data 240 } 241 ) 242 243 def write_compacted_month_if_updated(region: str, flavor: str, year: int, month: int, data: List[Tuple[datetime.datetime, int]]) -> bool: 244 current_compacted_data = sorted(retrieve_compacted_prices(region, flavor, year, month).items()) 245 calculated_data: List[Tuple[str, Decimal]]= [] 246 for item in data: 247 calculated_data.append((str(int(item[0].timestamp())), Decimal(item[1]))) 248 249 if all(item in current_compacted_data for item in calculated_data) and all(item in calculated_data for item in current_compacted_data): 250 print(f"{region} {flavor} {year} {month} No compaction update needed") 251 return False 252 else: 253 print(f"{region} {flavor} {year} {month} Compaction needed") 254 write_compacted_month(region, flavor, year, month, data) 255 return True 256 257 258 def retrieve_compacted_prices(region, flavor, year, month) -> dict: 259 table = DYNAMO_CLIENT.Table('wow-token-compacted') 260 pk = f'{region}-{flavor}-{year}-{month}' 261 response = table.query( 262 KeyConditionExpression=Key('region-flavor-timestamp').eq(pk) 263 ) 264 if response['Items']: 265 return response['Items'][0]['data'] 266 return {} 267 268 269 def _is_recent_month(current_time: datetime.datetime, given_time: datetime.datetime) -> bool: 270 difference = abs(current_time - given_time) 271 return difference.days <= 93 272 273 274 def _is_current_month(current_time: datetime.datetime, given_time: datetime.datetime) -> bool: 275 return current_time.month == given_time.month and current_time.year == given_time.year 276 277 278 def monthly_compact(region, flavor, year, month, current_time) -> None: 279 compacted_prices = retrieve_compacted_prices(region, flavor, year, month) 280 given_time = datetime.datetime(year, month, 1, tzinfo=datetime.timezone.utc) 281 recent = _is_recent_month(current_time, given_time) 282 283 if not compacted_prices and not recent: 284 prices = retrieve_prices_from_s3(region, flavor) 285 elif _is_recent_month(current_time, given_time): 286 prices = retrieve_recent_prices(region, flavor) 287 else: 288 return 289 290 compacted_prices = [] 291 for price in prices: 292 if price[0].month == month and price[0].year == year: 293 compacted_prices.append(price) 294 if compacted_prices: 295 print(f'Writing compacted data for {region} {flavor} {year} {month}') 296 write_compacted_month_if_updated(region, flavor, year, month, compacted_prices) 297 298 299 def compactor(region, flavor, compact_all: bool = False): 300 if compact_all: 301 if flavor == 'retail': 302 start_date = datetime.datetime.fromisoformat('2020-11-15 00:00:01.000000000+00:00') 303 else: 304 start_date = datetime.datetime.fromisoformat('2023-05-23 00:00:01.000000000+00:00') 305 else: 306 start_date = datetime.datetime.now(tz=datetime.UTC) - datetime.timedelta(days=62) 307 308 _current_time = datetime.datetime.now(tz=datetime.UTC) 309 _date = start_date 310 _month = start_date.month 311 _year = start_date.year 312 if compact_all: 313 monthly_compact(region, flavor, _year, _month, _current_time) 314 while _date <= _current_time: 315 _current_time - datetime.timedelta(days=1) 316 if _month != _date.month or _year != _date.year: 317 _month = _date.month 318 _year = _date.year 319 time.sleep(1) 320 monthly_compact(region, flavor, _year, _month, _current_time) 321 322 _date += datetime.timedelta(days=1) 323 324 325 def lambda_handler(event, context): 326 regions = ['us', 'eu', 'tw', 'kr'] 327 flavors = ['retail', 'classic'] 328 times = ['30d', '90d', '6m', '1y', '2y', 'all'] 329 for region in regions: 330 for flavor in flavors: 331 history = PriceHistory(region, flavor) 332 compactor(region, flavor) 333 for t in times: 334 if not (flavor == 'classic' and t == '2y'): 335 history.write_binned_if_updated(t) 336 337 338 def main(): 339 #print(retrieve_prices_from_s3('us', 'retail')) 340 # retrieve_recent_prices('us', 'retail') 341 # retrieve_compacted_prices('us', 'retail', '2024', '1') 342 regions = ['us', 'eu', 'tw', 'kr'] 343 flavors = ['retail', 'classic'] 344 times = ['30d', '90d', '6m', '1y', '2y', 'all'] 345 for region in regions: 346 for flavor in flavors: 347 compactor(region, flavor) 348 history = PriceHistory(region, flavor) 349 #for t in times: 350 # if not (flavor == 'classic' and t == '2y'): 351 # history.write_binned_if_updated(t) 352 #history.write_binned_prices(t) 353 354 355 356 if __name__ == '__main__': 357 main()