pruncalc/market_stats.py

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()