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(f: typing.TextIO) -> typing.Mapping[str, float]:
	raw_prices: typing.Sequence[Price] = 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()}

	hardcoded_prices = {
		'AFP': 65638,
		'ANZ': 70601,
		'BFP': 23408,
		'BSU': 605062,
		'CPU': 1426423,
		'CRU': 171386,
		'DIS': 19583,
		'GCH': 18303,
		'GNZ': 30361,
		'GWS': 9778478,
		'HAM': 4686751,
		'HNZ': 93580,
		'IMM': 101522,
		'JUI': 0,
		'NOZ': 74525,
		'PFG': 2677222,
		'PK': 1144,
		'RDS': 598170,
		'SDM': 1721027,
		'SST': 5863587,
		'SU': 157860,
		'SUD': 84327,
		'TAC': 245797,
		'TOR': 540169,
		'VCB': 673713,
		'VFT': 1781416,
		'VOE': 3699358,
		'VOR': 2547315,
		'WR': 129412,
	}
	assert frozenset(prices).isdisjoint(hardcoded_prices), frozenset(prices).intersection(hardcoded_prices)
	prices.update(hardcoded_prices)
	return prices

def add_company_ranks(totals: dict[str, CompanyTotals]) -> dict[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: typing.Mapping[str, dict]) -> 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 CompanyTickerData(typing.TypedDict):
	amount: float
	volume: float
	rank: int

class CompanyTotals(typing.TypedDict, total=False):
	volume: float
	volumeRank: int