pruncalc/roi.py

from __future__ import annotations

import dataclasses
import json
import typing

import cache

def main() -> None:
	recipes: list[Recipe] = cache.get('https://api.prunplanner.org/data/recipes/')
	buildings: dict[str, Building] = {m['building_ticker']: m for m in cache.get('https://api.prunplanner.org/data/buildings/')}
	materials: dict[str, Material] = {m['ticker']: m for m in cache.get('https://api.prunplanner.org/data/materials/')}
	raw_prices: list[RawPrice] = cache.get('https://refined-prun.github.io/refined-prices/all.json')
	for cx in ['AI1', 'CI1', 'IC1', 'NC1']:
		profits = calc_for_cx(cx, recipes, buildings, materials, raw_prices)
		with open(f'www/roi_{cx.lower()}.json', 'w') as f:
			json.dump([dataclasses.asdict(p) for p in profits], f, indent='\t')

def calc_for_cx(cx: str, recipes: typing.Collection[Recipe], buildings: typing.Mapping[str, Building],
		materials: typing.Mapping[str, Material], raw_prices: typing.Collection[RawPrice]) -> typing.Sequence[Profit]:
	prices: dict[str, Price] = {
		p['MaterialTicker']: Price(p['VWAP7D'], p['AverageTraded7D'], p['VWAP30D']) for p in raw_prices # pyright: ignore[reportArgumentType]
		if p['ExchangeCode'] == cx
	}
	habitation: typing.Mapping[Worker, str] = {
		'pioneers': 'HB1',
		'settlers': 'HB2',
		'technicians': 'HB3',
		'engineers': 'HB4',
		'scientists': 'HB5',
	}
	hab_area_cost: dict[Worker, float] = {}
	hab_capex: dict[Worker, float] = {}
	for worker, hab in habitation.items():
		hab_area_cost[worker] = buildings[hab]['area_cost'] / 100
		hab_capex[worker] = building_construction_cost(buildings[hab], prices) / 100

	profits: list[Profit] = []
	for recipe in recipes:
		if profit := calc_profit(recipe, buildings, hab_area_cost, hab_capex, materials, prices):
			profits.append(profit)
	profits.sort()
	return profits

def calc_profit(recipe: Recipe, buildings: typing.Mapping[str, Building], hab_area_cost: typing.Mapping[Worker, float],
		hab_capex: typing.Mapping[Worker, float], materials: typing.Mapping[str, Material],
		prices: typing.Mapping[str, Price]) -> Profit | None:
	if len(recipe['outputs']) == 0:
		return

	outputs: list[MatPrice] = []
	revenue = 0
	output_prices: dict[str, PriceNonNull] = {}
	for output in recipe['outputs']:
		price = prices[output['material_ticker']]
		if price.vwap_7d is None or price.average_traded_7d is None:
			return # skip recipes with thinly traded outputs
		output_prices[output['material_ticker']] = typing.cast(PriceNonNull, price)
		outputs.append(MatPrice(output['material_ticker'], output['material_amount'], price.vwap_7d))
		revenue += price.vwap_7d * output['material_amount']

	input_costs: list[MatPrice] = []
	cost = 0
	for input in recipe['inputs']:
		if (input_cost := prices[input['material_ticker']].vwap_7d) is None:
			return # skip recipes with thinly traded inputs
		input_costs.append(MatPrice(input['material_ticker'], input['material_amount'], input_cost))
		cost += input_cost * input['material_amount']
	profit_per_run = revenue - cost

	building = buildings[recipe['building_ticker']]
	area = building['area_cost'] + sum(hab_area_cost[worker] * building[worker] for worker in hab_area_cost)
	capex = building_construction_cost(building, prices) + \
		sum(hab_capex[worker] * building[worker] for worker in hab_capex)
	runs_per_day = 24 * 60 * 60 * 1000 / recipe['time_ms']
	if building['building_ticker'] in ('FRM', 'ORC'):
		runs_per_day *= 1.1212 # promitor's fertility
	worker_consumable_daily_cost = building_daily_cost(building, prices)
	cost_per_day = cost * runs_per_day + worker_consumable_daily_cost

	lowest_liquidity = min(recipe['outputs'],
			key=lambda output: output['material_amount'] / output_prices[output['material_ticker']].average_traded_7d)
	output_per_day = lowest_liquidity['material_amount'] * runs_per_day

	logistics_per_area = max(
		sum(materials[input['material_ticker']]['weight'] * input['material_amount'] for input in recipe['inputs']),
		sum(materials[input['material_ticker']]['volume'] * input['material_amount'] for input in recipe['inputs']),
		sum(materials[output['material_ticker']]['weight'] * output['material_amount'] for output in recipe['outputs']),
		sum(materials[output['material_ticker']]['volume'] * output['material_amount'] for output in recipe['outputs']),
	) * runs_per_day / area
	return Profit(outputs, recipe['recipe_name'],
			expertise=building['expertise'],
			profit_per_day=(profit_per_run * runs_per_day - worker_consumable_daily_cost),
			area=area,
			capex=capex,
			cost_per_day=cost_per_day,
			input_costs=input_costs,
			worker_consumable_cost_per_day=worker_consumable_daily_cost,
			runs_per_day=runs_per_day,
			logistics_per_area=logistics_per_area,
			output_per_day=output_per_day,
			average_traded_7d=output_prices[lowest_liquidity['material_ticker']].average_traded_7d)

def building_construction_cost(building: Building, prices: typing.Mapping[str, Price]) -> float:
	return sum(bc['material_amount'] * prices[bc['material_ticker']].vwap_7d for bc in building['costs']) # pyright: ignore[reportOperatorIssue]

def building_daily_cost(building: Building, prices: typing.Mapping[str, Price]) -> float:
	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.1)],
	}
	cost = 0
	for worker, mats in consumption.items():
		workers = building[worker]
		for mat, per_100 in mats:
			mat_price = prices[mat]
			cost += (mat_price.vwap_7d or mat_price.vwap_30d) * workers * per_100 / 100
	return cost

Worker = typing.Literal['pioneers', 'settlers', 'technicians', 'engineers', 'scientists']

class Recipe(typing.TypedDict):
	recipe_name: str
	building_ticker: str
	inputs: list[RecipeMat]
	outputs: list[RecipeMat]
	time_ms: int

class RecipeMat(typing.TypedDict):
	material_ticker: str
	material_amount: int

class Building(typing.TypedDict):
	building_ticker: str
	expertise: str
	area_cost: int
	costs: list[BuildingMat]
	pioneers: int
	settlers: int
	technicians: int
	engineers: int
	scientists: int

class BuildingMat(typing.TypedDict):
	material_ticker: str
	material_amount: int

class Material(typing.TypedDict):
	ticker: str
	weight: float
	volume: float

class RawPrice(typing.TypedDict):
	MaterialTicker: str
	ExchangeCode: str
	VWAP7D: float | None # volume-weighted average price over last 7 days
	AverageTraded7D: float | None # averaged daily traded volume over last 7 days
	VWAP30D: float | None

@dataclasses.dataclass(eq=False, frozen=True, slots=True)
class Price:
	vwap_7d: float | None
	average_traded_7d: float | None
	vwap_30d: float | None

@dataclasses.dataclass(eq=False, frozen=True, slots=True)
class PriceNonNull:
	vwap_7d: float
	average_traded_7d: float

@dataclasses.dataclass(eq=False, frozen=True, slots=True)
class Profit:
	outputs: typing.Collection[MatPrice]
	recipe: str
	expertise: str
	profit_per_day: float
	area: float
	capex: float
	cost_per_day: float
	input_costs: typing.Collection[MatPrice]
	worker_consumable_cost_per_day: float
	runs_per_day: float
	logistics_per_area: float
	output_per_day: float
	average_traded_7d: float

	def __lt__(self, other: Profit) -> bool:
		if (break_even := self.capex / self.profit_per_day) < 0:
			break_even = 10000 - self.profit_per_day
		if (other_break_even := other.capex / other.profit_per_day) < 0:
			other_break_even = 10000 - other.profit_per_day
		return break_even < other_break_even

@dataclasses.dataclass(eq=False, frozen=True, slots=True)
class MatPrice:
	ticker: str
	amount: int
	vwap_7d: float

if __name__ == '__main__':
	main()