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)

	markets_with_trades = ranking = ic1_lowest = gap = 0
	for mat, mat_prices in markets.items():
		mat_prices.sort(key=lambda p: (p['Traded7D'] or 0))
		if mat_prices[0]['Traded7D'] == None:
			continue
		markets_with_trades += 1
		for index, price in enumerate(mat_prices):
			if price['ExchangeCode'] == 'IC1':
				break
		else:
			raise AssertionError('IC1 not found for ' + mat)
		ranking += index
		if index == 0:
			ic1_lowest += 1
			gap += (mat_prices[1]['Traded7D'] - mat_prices[0]['Traded7D']) / mat_prices[1]['Traded7D'] # type: ignore
	print(f'IC1 ranking {ranking}, lowest in {ic1_lowest} of {markets_with_trades} markets, gap {gap:.2f}')

if __name__ == '__main__':
	main()