|
|
@@ -0,0 +1,55 @@
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import collections
|
|
|
+import datetime
|
|
|
+import json
|
|
|
+import typing
|
|
|
+
|
|
|
+import dulwich.repo
|
|
|
+import dulwich.objects
|
|
|
+
|
|
|
+if typing.TYPE_CHECKING:
|
|
|
+ import market
|
|
|
+
|
|
|
+def main() -> None:
|
|
|
+ repo = dulwich.repo.Repo('../refined-prices')
|
|
|
+
|
|
|
+ today = datetime.datetime.now(tz=datetime.UTC).replace(hour=0, minute=0, second=0, microsecond=0)
|
|
|
+ sunday = (today - datetime.timedelta(days=(today.weekday() + 1) % 7))
|
|
|
+ for _ in range(52):
|
|
|
+ analyze_markets(prices_on_day(repo, sunday))
|
|
|
+ sunday -= datetime.timedelta(days=7)
|
|
|
+
|
|
|
+def prices_on_day(repo: dulwich.repo.Repo, day: datetime.datetime) -> typing.Sequence[market.RawPrice]:
|
|
|
+ '''refined-prices for the earliest commit on the given day'''
|
|
|
+ day_ts = int(day.timestamp())
|
|
|
+ next_day_ts = int((day + datetime.timedelta(days=1)).timestamp())
|
|
|
+ *_, entry = repo.get_walker(paths=[b'all.json'], since=day_ts, until=next_day_ts)
|
|
|
+ dt = datetime.datetime.fromtimestamp(entry.commit.author_time, tz=datetime.UTC)
|
|
|
+ print('loading refined-prices', dt, entry.commit.tree.decode())
|
|
|
+ tree = typing.cast(dulwich.objects.Tree, repo[entry.commit.tree])
|
|
|
+ _, blob = tree[b'all.json']
|
|
|
+ return json.loads(typing.cast(dulwich.objects.Blob, repo[blob]).data.decode())
|
|
|
+
|
|
|
+def analyze_markets(prices: typing.Sequence[market.RawPrice]) -> None:
|
|
|
+ markets: dict[str, list[market.RawPrice]] = collections.defaultdict(list)
|
|
|
+ for price in prices:
|
|
|
+ if price['ExchangeCode'].endswith('2'):
|
|
|
+ continue
|
|
|
+ markets[price['MaterialTicker']].append(price)
|
|
|
+
|
|
|
+ ic1_lowest = 0
|
|
|
+ for mat_prices in markets.values():
|
|
|
+ (ic1_price,) = (price for price in mat_prices if price['ExchangeCode'] == 'IC1')
|
|
|
+ lowest_traded = 1_000_000
|
|
|
+ for price in mat_prices:
|
|
|
+ if price['ExchangeCode'] == 'IC1':
|
|
|
+ continue
|
|
|
+ if price['Traded7D'] is None or price['Traded7D'] < lowest_traded:
|
|
|
+ lowest_traded = price['Traded7D'] or 0
|
|
|
+ if (ic1_price['Traded7D'] or 0) < lowest_traded:
|
|
|
+ ic1_lowest += 1
|
|
|
+ print('IC1 lowest', ic1_lowest, 'of', len(markets))
|
|
|
+
|
|
|
+if __name__ == '__main__':
|
|
|
+ main()
|
