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 calc_consumption_profit(month: str, prices: typing.Mapping[str, float],
		prod_data: typing.Mapping[str, ProdData], company_data: CompanyData) -> None:
	consumption_ratios = get_consumption_ratios(month)
	# 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)
	universe_profit: dict[str, float] = collections.defaultdict(float)
	for co_id, co_data in company_data['individual'].items():
		co_profit = 0.0
		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
			elif (consumption_ratio := consumption_ratios.get(mat)) is not None:
				unit_cost = 0
				for consumed_mat, ratio in consumption_ratio.items():
					unit_cost += ratio * prices[consumed_mat]
					universe_consumption[consumed_mat] += ratio * mat_data['amount']
				mat_data['profit'] = mat_data['volume'] - unit_cost * mat_data['amount']
			else:
				raise AssertionError('no cost for producing ' + mat)
			co_profit += mat_data['profit']
			universe_profit[mat] += mat_data['profit']
		company_data['totals'][co_id]['profit'] = co_profit

	add_company_ranks(company_data['totals'])

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

def get_consumption_ratios(month: str) -> typing.Mapping[str, typing.Mapping[str, float]]:
	with open(f'rawData/{month}-recipes.json', 'r') as f:
		recipes: typing.Mapping[str, Recipe] = {recipe['RecipeName']: recipe for recipe in json.load(f)}
	with open(f'rawData/{month}-recipe-ratios.json', 'r') as f:
		recipe_ratios: dict[str, typing.Mapping[str, float]] = json.load(f)
	with open('rawData/buildings.json', 'r') as f:
		buildings: typing.Mapping[str, typing.Mapping] = {b['Ticker']: b for b in json.load(f)}

	outputs: dict[str, typing.Mapping[str, float]] = {}
	duplicates = set() # mats that can be produced by more than 1 recipe
	for recipe in recipes.values():
		for output in recipe['Outputs']:
			output_ticker = output['Ticker']
			if output_ticker in outputs:
				duplicates.add(output_ticker)
				continue
			ratios = consumption_ratio_for_recipe(recipe, output, buildings)
			outputs[output_ticker] = ratios
	for dupe in duplicates:
		recipe_ratio = recipe_ratios.pop(dupe)
		assert dupe == 'SIO' or abs(sum(recipe_ratio.values()) - 1.0) < 1e-6, f"recipe ratio for {dupe} doesn't sum to 1"
		ratios = collections.defaultdict(float)
		for recipe_name, weight in recipe_ratio.items():
			recipe = recipes[recipe_name]
			(output,) = (o for o in recipe['Outputs'] if o['Ticker'] == dupe)
			for input_item, ratio in consumption_ratio_for_recipe(recipe, output, buildings).items():
				ratios[input_item] += weight * ratio
		outputs[dupe] = ratios
	assert len(recipe_ratios) == 0, f'unused recipe ratios: {recipe_ratios.keys()}'
	return outputs

worker_daily_consumption = {
	'Pioneers': [('COF', 0.5), ('DW', 4), ('RAT', 4), ('OVE', 0.5), ('PWO', 0.2)],
	'Settlers': [('DW', 5), ('RAT', 6), ('KOM', 1), ('EXO', 0.5), ('REP', 0.2), ('PT', 0.5)],
	'Technicians': [('DW', 7.5), ('RAT', 7), ('ALE', 1), ('MED', 0.5), ('SC', 0.1), ('HMS', 0.5), ('SCN', 0.1)],
	'Engineers': [('DW', 10), ('MED', 0.5), ('GIN', 1), ('FIM', 7), ('VG', 0.2), ('HSS', 0.2), ('PDA', 0.1)],
	'Scientists': [('DW', 10), ('MED', 0.5), ('WIN', 1), ('MEA', 7), ('NST', 0.1), ('LC', 0.2), ('WS', 0.05)],
}
def consumption_ratio_for_recipe(recipe: Recipe, output: RecipeMat, buildings: dict) -> typing.Mapping[str, float]:
	ratios = {
		input['Ticker']: input['Amount'] / output['Amount'] for input in recipe['Inputs']
	}
	building = buildings[recipe['BuildingTicker']]
	for worker, mats in worker_daily_consumption.items():
		workers = building[worker]
		for mat, per_100 in mats:
			consumed_per_day = workers * per_100 / 100
			# assume 160% efficiency
			consumed_per_unit_output = consumed_per_day / output['Amount'] * (recipe['TimeMs'] / (24*60*60*1000) / 1.6)
			ratios[mat] = ratios.get(mat, 0) + consumed_per_unit_output
	return ratios

def add_company_ranks(totals: typing.Mapping[str, CompanyTotals]) -> None:
	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
	ranked = sorted(totals.items(), key=lambda item: item[1]['profit'], reverse=True) # pyright: ignore[reportTypedDictNotRequiredAccess]
	for rank, (company_id, company_totals) in enumerate(ranked, start=1):
		company_totals['profitRank'] = rank

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': sum(mat['profit'] for mat in prod_data.values()), # pyright: ignore[reportTypedDictNotRequiredAccess]
			'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 Recipe(typing.TypedDict):
	BuildingTicker: str
	RecipeName: str
	Inputs: typing.Sequence[RecipeMat]
	Outputs: typing.Sequence[RecipeMat]
	TimeMs: int

class RecipeMat(typing.TypedDict):
	Ticker: str
	Amount: float

class ProdData(typing.TypedDict):
	amount: float
	volume: float
	consumed: typing.NotRequired[float]
	profit: 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
	profit: typing.NotRequired[float]
	volumeRank: typing.NotRequired[int]
	profitRank: typing.NotRequired[int]