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 calc_consumption_profit(prod_data: typing.Mapping[str, ProdData], company_data: CompanyData) -> None:
	# reddead_solo derived these from apr26 data
	extraction_cost = {
		'ALO': 46.469626236277556,
		'AMM': 64.22413645940814,
		'AR': 47.70764360662854,
		'AUO': 54.44708670483523,
		'BER': 53.972209698989495,
		'BOR': 83.72543770145344,
		'BRM': 27.888018032559867,
		'BTS': 78.06705317448308,
		'CLI': 35.136976457308684,
		'CUO': 50.86203030902945,
		'F': 131.4747740048254,
		'FEO': 48.650761183540155,
		'GAL': 57.92699044231113,
		'H': 22.44162253371223,
		'H2O': 20.786470813001372,
		'HAL': 73.71971131519504,
		'HE': 75.22625975185184,
		'HE3': 124.06892211329114,
		'HEX': 42.936879245965734,
		'KR': 69.26330139617842,
		'LES': 107.34219811491415,
		'LIO': 27.095584832287784,
		'LST': 22.711465634512184,
		'MAG': 187.27617420211016,
		'MGS': 58.51507605569296,
		'N': 27.282314763847143,
		'NE': 59.13606618175083,
		'O': 27.57488101890798,
		'REO': 80.67539173234522,
		'SCR': 55.37980023917345,
		'SIO': 31.748301836118316,
		'TAI': 54.2528323296315,
		'TCO': 67.4922174225306,
		'TIO': 40.709175823948684,
		'TS': 47.68370385324906,
		'ZIR': 53.627181874429766,
	}
	pio_consumable_price_ratios = {
		'DW': 0.23852885030544477 / 107.1798998000011,
		'RAT': 0.37223040985862066 / 167.25699210000184,
		'OVE': 0.03871106977501096 / 139.15460790000103,
		'COF': 0.302139841269314 / 1086.1015050000133,
		'PWO': 0.04838982879160972 / 434.86706069999866,
	}

	universe_consumption: dict[str, float] = collections.defaultdict(float)
	for co_data in company_data['individual'].values():
		for mat, mat_data in co_data.items():
			if (cost := extraction_cost.get(mat)) is not None:
				total_cost = mat_data['amount'] * cost
				mat_data['profit'] = mat_data['volume'] - total_cost
				for consumed_mat, ratio in pio_consumable_price_ratios.items():
					universe_consumption[consumed_mat] += ratio * total_cost

	for mat, mat_data in prod_data.items():
		mat_data['consumed'] = universe_consumption.get(mat, 0.0)

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
	consumed: typing.NotRequired[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
	profit: typing.NotRequired[float]
	rank: int

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