growing.py

@@ -0,0 +1,52 @@
+import collections
+
+import httpx
+
+def main():
+	corps = httpx.get('https://prun.raylu.net/stats/data/parentCorps.json').json()
+	companies = httpx.get('https://prun.raylu.net/stats/data/knownCompanies.json').json()
+
+	volume_delta(corps, companies)
+
+def volume_delta(corps, companies):
+	co_totals = httpx.get('https://prun.raylu.net/stats/data/company-data-jul26.json').json()['totals']
+	jul_volume = month_volume(corps, companies, co_totals)
+	co_totals = httpx.get('https://prun.raylu.net/stats/data/company-data-aug26.json').json()['totals']
+	aug_volume = month_volume(corps, companies, co_totals)
+
+	for corp, volume in aug_volume.items():
+		print(corp, volume - jul_volume.get(corp, 0))
+
+def month_volume(corps, companies, co_totals) -> dict:
+	volumes = collections.defaultdict(int)
+	for coid, pd in co_totals.items():
+		co = companies.get(coid)
+		if co is None: continue
+		corp = co.get('Corporation')
+		if corp is None: continue
+		corp = corps.get(corp, corp)
+		volumes[corp] += pd['volume']
+	return volumes
+
+def base_delta(corps, companies):
+	base_data = httpx.get('https://prun.raylu.net/stats/data/base-data-jul26.json').json()
+	jul_bases = month_bases(corps, companies, base_data)
+	base_data = httpx.get('https://prun.raylu.net/stats/data/base-data-aug26.json').json()
+	aug_bases = month_bases(corps, companies, base_data)
+
+	for corp, base_count in aug_bases.items():
+		print(corp, base_count - jul_bases.get(corp, 0))
+
+def month_bases(corps, companies, base_data) -> dict:
+	base_counts = collections.defaultdict(int)
+	for coid, bd in base_data.items():
+		co = companies.get(coid)
+		if co is None: continue
+		corp = co.get('Corporation')
+		if corp is None: continue
+		corp = corps.get(corp, corp)
+		base_counts[corp] += bd['bases']
+	return base_counts
+
+if __name__ == '__main__':
+	main()

market.py

@@ -250,6 +250,7 @@ class RawPrice(typing.TypedDict):
 	HighYesterday: float | None
 	LowYesterday: float | None
 	AverageTraded7D: float | None # averaged daily traded volume over last 7 days
+	Traded7D: int | None
 	Traded30D: int | None
 
 class PriceChartPoint(typing.TypedDict):

market_stats.py

@@ -0,0 +1,114 @@
+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()

newbies.py

@@ -1,26 +1,43 @@
 from __future__ import annotations
 
 import datetime
+import itertools
 import sys
 import typing
 
+import httpx
+
 import cache
 
 def main() -> None:
+	since = sys.argv[1] + 'T00:00:00Z'
+
+	codes: dict[str, str] = {company['UserName'].casefold(): company['CompanyCode']
+			for company in cache.get('https://rest.fnar.net/company/all') if company['UserName'] is not None}
+
 	planets = ['Avalon', 'Berthier', 'Boucher', 'Nova Honshu', 'Promitor']
 	joined: dict[str, int] = {}
 	for planet in planets:
 		print('querying', planet, 'chat...', file=sys.stderr)
-		messages: typing.Sequence[Message] = cache.get(f'https://api.fnar.net/chat/messages?channel_names={planet} Global Site Owners')
+		messages: typing.Sequence[Message] = httpx.get('https://api.fnar.net/chat/messages',
+				params={'channel_names': planet + ' Global Site Owners', 'updated_since': since}).raise_for_status().json()
 		for message in messages:
 			if message['Type'] == 'JOINED':
 				old_join = joined.get(message['SenderUserName'])
 				if old_join is None or old_join > message['MessageTimestamp']:
 					joined[message['SenderUserName']] = message['MessageTimestamp']
 
-	for user_id, join_ts in sorted(joined.items(), key=lambda i: i[1]):
+	relevant_cos = {co for username in joined.keys() if (co := codes.get(username))}
+	rep: dict[str, int] = {}
+	for co_batch in itertools.batched(relevant_cos, 10):
+		companies = httpx.get('https://api.fnar.net/company/lookup?' + '&'.join(f'company={co}' for co in co_batch)).raise_for_status().json()
+		for company in companies:
+			if company is not None:
+				rep[company['UserName'].casefold()] = company['Reputation']
+	
+	for username, join_ts in sorted(joined.items(), key=lambda i: i[1]):
 		join_str = datetime.datetime.fromtimestamp(join_ts/1000, tz=datetime.UTC).strftime('%Y-%m-%d %H:%M:%S')
-		print(user_id, join_str, sep='\t')
+		print(username, join_str, '', '', rep.get(username, ''), sep='\t')
 
 class Message(typing.TypedDict):
 	Type: typing.Literal['JOINED', 'LEFT', 'MESSAGE']

pyproject.toml

@@ -3,10 +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]
@@ -15,5 +18,6 @@ dev = ['workers-py', 'workers-runtime-sdk']
 [tool.ruff.lint]
 ignore = [
 	'I001', # unsorted-imports
+	'SIM118', #in-dict-keys
 	'TRY002', # raise-vanilla-class
 ]