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:
	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(52):
		weekly_stats.append(analyze_markets(prices_on_day(repo, sunday), sunday.date()))
		sunday -= datetime.timedelta(days=7)
	render_chart(list(reversed(weekly_stats)))

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

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

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 = 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}')
	return {
		'date': date.isoformat(),
		'ranking': ranking,
		'ic1_lowest': ic1_lowest,
		'markets_with_trades': markets_with_trades,
		'gap': gap,
	}

def render_chart(weekly_stats: typing.Sequence[MarketStats]) -> None:
	base = altair.Chart({'values': list(weekly_stats)}).encode(
		x=altair.X('date:T', title='week'),
		tooltip=[
			altair.Tooltip('date:T', title='week'),
			altair.Tooltip('ranking:Q', title='ranking'),
			altair.Tooltip('ic1_lowest:Q', title='IC1 lowest'),
			altair.Tooltip('markets_with_trades:Q', title='markets with trades'),
			altair.Tooltip('gap:Q', title='gap', format='.2f'),
		],
	)
	chart = altair.vconcat(
		base.mark_line(point=True).encode(y=altair.Y('ranking:Q', title='sum of IC1 rank')).properties(height=160),
		base.mark_line(point=True).encode(y=altair.Y('ic1_lowest:Q', title='# markets where IC1 is lowest')).properties(height=160),
		base.mark_line(point=True).encode(y=altair.Y('gap:Q', title='trade gap')).properties(height=160),
		title='IC1 market stats',
	)
	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()

if __name__ == '__main__':
	main()