market: render chart with altair and kitty graphics protocol

market_stats.py

@@ -1,10 +1,14 @@
 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
 
@@ -16,9 +20,11 @@ def main() -> None:
 
 	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):
-		analyze_markets(prices_on_day(repo, sunday))
+		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'''
@@ -31,7 +37,14 @@ def prices_on_day(repo: dulwich.repo.Repo, day: datetime.datetime) -> typing.Seq
 	_, 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:
+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'):
@@ -54,6 +67,48 @@ def analyze_markets(prices: typing.Sequence[market.RawPrice]) -> None:
 			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()

pyproject.toml

@@ -3,11 +3,13 @@ name = 'pruncalc'
 version = '0'
 requires-python = '>=3.13'
 dependencies = [
+	'altair',
 	'cbor2',
 	'dulwich',
 	'h2',
 	'httpx',
 	'typed-argument-parser',
+	'vl-convert-python', # altair PNG output
 ]
 
 [dependency-groups]