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['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]['AreaCost'] / 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['Ticker']]
		if price.vwap_7d is None or price.average_traded_7d is None:
			return # skip recipes with thinly traded outputs
		output_prices[output['Ticker']] = typing.cast(PriceNonNull, price)
		outputs.append(MatPrice(output['Ticker'], output['Amount'], price.vwap_7d))
		revenue += price.vwap_7d * output['Amount']

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

	building = buildings[recipe['BuildingTicker']]
	area = building['AreaCost'] + 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['TimeMs']
	if 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['Amount'] / output_prices[output['Ticker']].average_traded_7d)
	output_per_day = lowest_liquidity['Amount'] * runs_per_day

	logistics_per_area = max(
		sum(materials[input['Ticker']]['Weight'] * input['Amount'] for input in recipe['Inputs']),
		sum(materials[input['Ticker']]['Volume'] * input['Amount'] for input in recipe['Inputs']),
		sum(materials[output['Ticker']]['Weight'] * output['Amount'] for output in recipe['Outputs']),
		sum(materials[output['Ticker']]['Volume'] * output['Amount'] for output in recipe['Outputs']),
	) * runs_per_day / area
	return Profit(outputs, recipe['RecipeName'],
			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['Ticker']].average_traded_7d)

def building_construction_cost(building: Building, prices: typing.Mapping[str, Price]) -> float:
	return sum(bc['Amount'] * prices[bc['CommodityTicker']].vwap_7d for bc in building['BuildingCosts']) # 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):
	RecipeName: str
	BuildingTicker: str
	Inputs: list[RecipeMat]
	Outputs: list[RecipeMat]
	TimeMs: int

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

class Building(typing.TypedDict):
	Ticker: str
	Expertise: str
	AreaCost: int
	BuildingCosts: list[BuildingMat]
	Pioneers: int
	Settlers: int
	Technicians: int
	Engineers: int
	Scientists: int

class BuildingMat(typing.TypedDict):
	CommodityTicker: str
	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()