prunstats/py/prepare.py

from __future__ import annotations
import httpx

import collections
import dataclasses
import csv
import json
import typing

def read_data(f: typing.TextIO) -> dict[str, list[Row]]:
	data: dict[str, list[Row]] = collections.defaultdict(list)
	reader = csv.reader(f)
	for row in reader:
		data[row[0]].append(Row(int(row[1]), int(row[2]), row[3]))
	return data

def get_prices(month: str) -> typing.Mapping[str, float]:
	with open(f'rawData/{month}-prices.json', 'r') as f:
		raw_prices: typing.Sequence[Price] = json.load(f)
	with open(f'rawData/{month}-prices-hardcoded.json', 'r') as f:
		hardcoded_prices: typing.Mapping[str, int] = json.load(f)
	volumes: dict[str, float] = collections.defaultdict(float)
	traded: dict[str, int] = collections.defaultdict(int)
	for price in raw_prices:
		if price['Traded30D'] is None:
			continue
		assert price['VWAP30D'] is not None
		volumes[price['MaterialTicker']] += price['VWAP30D'] * price['Traded30D']
		traded[price['MaterialTicker']] += price['Traded30D']

	prices = {ticker: volume / traded[ticker] for ticker, volume in volumes.items()}
	assert frozenset(prices).isdisjoint(hardcoded_prices), frozenset(prices).intersection(hardcoded_prices)
	prices.update(hardcoded_prices)
	return prices

def add_company_ranks(totals: typing.Mapping[str, CompanyTotals]) -> typing.Mapping[str, CompanyTotals]:
	ranked = sorted(totals.items(), key=lambda item: item[1]['volume'], reverse=True)
	for rank, (company_id, company_totals) in enumerate(ranked, start=1):
		company_totals['volumeRank'] = rank
	return totals

def write_data(month: str, data: dict[str, list[Row]],
		prod_data: typing.Mapping[str, ProdData], company_data: CompanyData) -> None:
	with open(f'www/data/prod-data-{month}.json', 'w') as f:
		json.dump(prod_data, f)
	with open(f'www/data/company-data-{month}.json', 'w') as f:
		json.dump(company_data, f)

	bases_data: dict[str, dict[str, int]] = {r.company_id: {'bases': r.num, 'rank': r.rank} for r in data['BASES']}
	ships_data: dict[str, dict[str, int]] = {r.company_id: {'ships': r.num, 'rank': r.rank} for r in data['SHIPS']}
	for company_id in company_data['totals']:
		bases_data.setdefault(company_id, {'bases': 1})
		ships_data.setdefault(company_id, {'ships': 2})
	with open(f'www/data/base-data-{month}.json', 'w') as f:
		json.dump(bases_data, f)
	with open(f'www/data/ship-data-{month}.json', 'w') as f:
		json.dump(ships_data, f)

	with open('www/data/universe-data.json', 'r+') as f:
		universe_data = json.load(f)
		universe_data[month] = {
			'volume': sum(mat['volume'] for mat in prod_data.values()),
			'profit': None,
			'bases': sum(co['bases'] for co in bases_data.values()),
			'companies': len(company_data['totals']),
		}
		f.seek(0)
		json.dump(universe_data, f)
		f.truncate()

def check_missing_tickers(prod_data: typing.Mapping[str, ProdData]) -> None:
	response = httpx.get('https://api.fnar.net/material').raise_for_status()
	tickers = frozenset(mat['Ticker'] for mat in response.json() if mat['Ticker'] != 'CMK')
	if missing := tickers - prod_data.keys():
		print('warning: missing production data for tickers', missing)

@dataclasses.dataclass(frozen=True, slots=True, eq=False)
class Row:
	rank: int
	num: int
	company_id: str

class Price(typing.TypedDict):
	MaterialTicker: str
	VWAP30D: float | None
	Traded30D: int | None

class ProdData(typing.TypedDict):
	amount: float
	volume: float

class CompanyData(typing.TypedDict):
	totals: typing.Mapping[str, CompanyTotals]
	individual: typing.Mapping[str, typing.Mapping[str, CompanyTickerData]]

class CompanyTickerData(typing.TypedDict):
	amount: float
	volume: float
	rank: int

class CompanyTotals(typing.TypedDict):
	volume: float
	volumeRank: typing.NotRequired[int]