math_path_handler.py (1962B)
1 import datetime 2 from typing import List, Tuple 3 4 from wow_token.db.cached_range import CachedRange 5 from wow_token.db.compacted import Compacted 6 from wow_token.db.recent import Recent 7 from wow_token.db.trinity import Trinity 8 from wow_token.flavor import Flavor 9 from wow_token.path_handler.relative_error import InvalidRelativePathError 10 from wow_token.path_handler.relative_path_handler import RelativePathHandler 11 from wow_token.region import Region 12 13 14 class MathPathHandler: 15 _cdb : Compacted 16 _rdb : Recent 17 def __init__(self, cdb: Compacted, rdb: Recent): 18 self._cdb = cdb 19 self._rdb = rdb 20 21 def path_handler(self, uri: str) -> List[Tuple[str, int]]: 22 # This URI takes the form of /v2/math/{math_function}/{flavor}/{region}/{range} 23 split_uri = uri.split('/') 24 math_function = split_uri[-4] 25 data = RelativePathHandler(self._cdb, self._rdb).path_handler(uri) 26 27 match math_function: 28 case 'avg': 29 return self._avg(data) 30 case _: 31 raise NotImplementedError 32 33 34 def _avg(self, data: List[Tuple[str, int]]) -> List[Tuple[str, int]]: 35 avg_buckets = [] 36 bucket_timestamp = None 37 bucket_price = 0 38 bucket_count = 0 39 for timestamp, price in data: 40 if bucket_timestamp is None: 41 bucket_timestamp = datetime.datetime.fromisoformat(timestamp) 42 elif bucket_timestamp.date() != datetime.datetime.fromisoformat(timestamp).date(): 43 bucket_head = datetime.datetime(year=bucket_timestamp.year, month=bucket_timestamp.month, day=bucket_timestamp.day) 44 avg_buckets.append((bucket_head.isoformat(), int(bucket_price/bucket_count))) 45 bucket_price = 0 46 bucket_count = 0 47 bucket_timestamp = datetime.datetime.fromisoformat(timestamp) 48 bucket_price += price 49 bucket_count += 1 50 return avg_buckets