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')
	prices: dict[str, Price] = {
		p['MaterialTicker']: Price(p['VWAP7D'], p['AverageTraded7D']) for p in raw_prices # pyright: ignore[reportArgumentType]
		if p['ExchangeCode'] == 'IC1' and p['VWAP7D'] is not None
	}

	profits: list[Profit] = []
	for recipe in recipes:
		if profit := calc_profit(recipe, buildings, materials, prices):
			profits.append(profit)
	profits.sort(reverse=True)
	print('\033[1mwrought    \033[0;32mdaily profit/area\033[31m')
	print('\033[30mrecipe                         \033[0mexpertise            \033[33mcapex \033[35mdaily opex \033[34mlogistics\033[0m')
	for p in profits:
		print(f'\033[53;1m{p.output:5} \033[53;32m{p.profit_per_area: 10,.0f} ', end='')
		warnings = []
		if p.low_volume:
			warnings.append('low volume')
		if p.logistics_per_area > 1.5:
			warnings.append('heavy logistics')
		if p.cost_per_day > 50_000:
			warnings.append('high opex')
		if len(warnings) > 0:
			print('              \033[31m' + ' '.join(warnings).ljust(43), end='')
		else:
			print(' ' * 14 + '\033[0;53m' + ' ' * 43, end='')
		print(f'\n\033[0;30m{p.recipe:30} \033[0m{p.expertise:19} \033[33m{p.capex:7,.0f} \033[35m{p.cost_per_day:9,.0f} \033[34m{p.logistics_per_area:5.2f}\033[0m')

	with open('www/roi.json', 'w') as f:
		json.dump([dataclasses.asdict(p) for p in profits], f, indent='\t')

def calc_profit(recipe: Recipe, buildings: typing.Mapping[str, Building], materials: typing.Mapping[str, Material],
		prices: typing.Mapping[str, Price]) -> Profit | None:
	try:
		(output,) = recipe['Outputs']
	except ValueError: # skip recipes that don't have exactly 1 output
		return
	try:
		output_price = prices[output['Ticker']]
		cost = sum(prices[input['Ticker']].vwap * input['Amount'] for input in recipe['Inputs'])
	except KeyError: # skip recipes with thinly traded materials
		return
	revenue = output_price.vwap * output['Amount']
	building = buildings[recipe['BuildingTicker']]
	capex = sum(bm['Amount'] * prices[bm['CommodityTicker']].vwap
			for bm in building['BuildingCosts'])
	profit_per_run = revenue - cost
	runs_per_day = 24 * 60 * 60 * 1000 / recipe['TimeMs']
	worker_consumable_daily_cost = building_daily_cost(building, prices)
	cost_per_day = cost * runs_per_day + worker_consumable_daily_cost
	output_per_day = output['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']),
		materials[output['Ticker']]['Weight'] * output['Amount'],
		materials[output['Ticker']]['Volume'] * output['Amount'],
	) * runs_per_day / building['AreaCost']
	return Profit(output['Ticker'], recipe['RecipeName'],
			expertise=building['Expertise'].replace('_', ' ').lower(),
			profit_per_area=(profit_per_run * runs_per_day - worker_consumable_daily_cost) / building['AreaCost'],
			capex=capex,
			cost_per_day=cost_per_day,
			logistics_per_area=logistics_per_area,
			low_volume=output_price.average_traded < output_per_day * 20)

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:
			cost += prices[mat].vwap * workers * per_100 / 100
	return cost

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
	PriceAverage: int
	VWAP7D: float | None # volume-weighted average price over last 7 days
	AverageTraded7D: float | None # averaged daily traded volume over last 7 days

@dataclasses.dataclass(eq=False, frozen=True, slots=True)
class Price:
	vwap: float
	average_traded: float

@dataclasses.dataclass(eq=False, slots=True)
class Profit:
	output: str
	recipe: str
	expertise: str
	profit_per_area: float
	capex: float
	cost_per_day: float
	logistics_per_area: float
	low_volume: bool
	score: float = dataclasses.field(init=False)

	def __post_init__(self) -> None:
		self.score = self.profit_per_area
		if self.low_volume:
			self.score *= 0.2

	def __lt__(self, other: Profit) -> bool:
		return self.score < other.score

if __name__ == '__main__':
	main()