pruncalc/market_stats.py

from __future__ import annotations

import collections
import base64
import datetime
import io
import json
import sys
import typing

import altair
import dulwich.repo
import dulwich.objects

if typing.TYPE_CHECKING:
	import market

def main() -> None:
	weeks = int(sys.argv[1])
	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))
	weekly_stats: list[MarketStats] = []
	for _ in range(weeks):
		weekly_stats.append(analyze_markets(prices_on_day(repo, sunday), sunday.date()))
		sunday -= datetime.timedelta(days=7)
	weekly_stats.reverse()

	base = altair.Chart({'values': weekly_stats}).encode(x=altair.X('date:T', title='week'))
	render_chart(base, 'ranking:Q', 'ranking', 'average IC1 rank')
	print('higher is better\naverage = (0 + 3) ÷ 2 = 1.5')
	render_chart(base, 'ic1_lowest:Q', '%', 'IC1 lowest')
	render_chart(base, 'ic1_zero:Q', '#', 'IC1 zero trades')
	render_chart(base, 'gap:Q', '%', 'trade gap')

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], date: datetime.date) -> MarketStats:
	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 = total_rank = ic1_lowest = ic1_zero = gap = 0
	for mat, mat_prices in markets.items():
		# find IC1's Traded7D rank among the CXes
		# group CXes by Traded7D first in case of ties
		rankings: dict[int, list[str]] = collections.defaultdict(list)
		for price in mat_prices:
			rankings[price['Traded7D'] or 0].append(price['ExchangeCode'])
		if len(rankings) == 1 and 0 in rankings: # no markets had any trades
			continue
		sorted_markets = sorted(rankings.items(), key=lambda item: item[0])
		rank = 0
		for _, cxes in sorted_markets:
			if 'IC1' in cxes:
				break
			rank += len(cxes)
		else:
			raise AssertionError('IC1 not found for ' + mat)
		markets_with_trades += 1
		total_rank += rank

		if rank == 0:
			ic1_lowest += 1
			ic1_traded = sorted_markets[0][0]
			if len(sorted_markets[0][1]) > 1: # another market is tied for last with us
				lowest_traded = ic1_traded
			else:
				lowest_traded = sorted_markets[1][0]
			if lowest_traded > 0:
				if ic1_traded == 0:
					ic1_zero += 1
				gap += (lowest_traded - ic1_traded) / lowest_traded
	print(f'{markets_with_trades} markets with trades, IC1 total rank {total_rank}, lowest in {ic1_lowest}, zero in {ic1_zero}, gap {gap:.2f}')
	return {
		'date': date.isoformat(),
		'ranking': total_rank / markets_with_trades,
		'ic1_lowest': ic1_lowest / markets_with_trades * 100,
		'ic1_zero': ic1_zero,
		'gap': gap / markets_with_trades * 100,
	}

def render_chart(base: altair.Chart, y: str, y_label: str, title: str) -> None:
	chart = base.mark_line(point=True).encode(y=altair.Y(y, title=y_label)).properties(title=title, height=160)
	buffer = io.BytesIO()
	chart.save(buffer, format='png')
	display_kitty_png(buffer.getvalue())

def display_kitty_png(png: bytes) -> None:
	encoded = base64.b64encode(png).decode()
	chunk_size = 4096
	for offset in range(0, len(encoded), chunk_size):
		part = encoded[offset:offset + chunk_size]
		more = int(offset + chunk_size < len(encoded))
		if offset == 0:
			params = f'a=T,f=100,q=2,m={more}'
		else:
			params = f'm={more}'
		sys.stdout.write(f'\033_G{params};{part}\033\\')
	sys.stdout.write('\n')
	sys.stdout.flush()

class MarketStats(typing.TypedDict):
	date: str
	ranking: float
	ic1_lowest: float
	ic1_zero: int
	gap: float

if __name__ == '__main__':
	main()