pruncalc/roi.py

from __future__ import annotations

import dataclasses
import typing

import cache

def main() -> None:
	recipes: list[Recipe] = cache.get('https://api.prunplanner.org/data/recipes')
	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 = []
	for recipe in recipes:
		if profit := calc_profit(recipe, materials, prices):
			profits.append(profit)
	profits.sort(reverse=True)
	for p in profits:
		print(f'{p.recipe:40} {p.profit:10.2f}\033[31m', end='')
		if p.low_volume:
			print(' low volume', end='')
		if p.high_opex:
			print(' high opex', end='')
		if p.heavy_logistics:
			print(' heavy logistics', end='')
		print('\033[0m')

def calc_profit(recipe: Recipe, 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']
	profit_per_run = revenue - cost
	runs_per_day = 24 * 60 * 60 * 1000 / recipe['TimeMs']
	cost_per_day = cost * runs_per_day
	output_per_day = output['Amount'] * runs_per_day
	logistics_per_day = 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
	return Profit(recipe['RecipeName'], profit=profit_per_run * runs_per_day,
			low_volume=output_price.average_traded < output_per_day * 20,
			high_opex=cost_per_day > 100_000,
			heavy_logistics=logistics_per_day > 100)

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 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:
	recipe: str
	profit: float
	low_volume: bool
	high_opex: bool
	heavy_logistics: bool
	score: float = dataclasses.field(init=False)

	def __post_init__(self) -> None:
		self.score = self.profit
		if self.low_volume:
			self.score *= 0.2
		if self.high_opex:
			self.score *= 0.8
		if self.heavy_logistics:
			self.score *= 0.5

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

if __name__ == '__main__':
	main()