pruncalc/market.py

from __future__ import annotations

import collections
import dataclasses
import sys
import typing

import cache
from config import config

def main() -> None:
	raw_prices: list[RawPrice] = cache.get('https://refined-prun.github.io/refined-prices/all.json')

	if len(sys.argv) > 1:
		exchange_tickers = sys.argv[1:]
		for ticker in exchange_tickers:
			(price,) = (p for p in raw_prices if p['FullTicker'] == ticker)
			a = analyze_price_chart(ticker, (price['Bid'] + price['Ask']) / 2) # pyright: ignore[reportOperatorIssue]
			print(f'{ticker}: bids filled = {a.bids_filled:6.0f}, asks filled = {a.asks_filled:6.0f}, profit per interval = {a.profit_per_interval:10.1f}')
		return

	check_cxos()

	markets: dict[str, list[Market]] = collections.defaultdict(list)
	for price in raw_prices:
		if (traded := price['AverageTraded7D']) is None or traded < 100:
			continue
		if price['Bid'] is None or price['Ask'] is None:
			continue
		if (high := price['HighYesterday']) is None or (low := price['LowYesterday']) is None:
			continue
		if (high - low) / high < 0.1:
			continue
		spread = (price['Ask'] - price['Bid']) / price['Ask']
		if spread < 0.15:
			continue
		chart_analysis = analyze_price_chart(price['FullTicker'], (price['Bid'] + price['Ask']) / 2)
		max_profit = (price['Ask'] - price['Bid']) * min(chart_analysis.bids_filled, chart_analysis.asks_filled)
		markets[price['ExchangeCode']].append(Market(price['FullTicker'], bid=price['Bid'], ask=price['Ask'],
				spread=spread, max_profit=max_profit, chart_analysis=chart_analysis))

	print(' mat       bid   ask spread  bids filled  asks filled  profit/time  max profit')
	for commodities in markets.values():
		commodities.sort(reverse=True)
		for m in commodities:
			print(f'{m.full_ticker:>8} {m.bid:5} {m.ask:5} {m.spread*100: 5.0f}% {m.chart_analysis.bids_filled:12.0f} '
					f'{m.chart_analysis.asks_filled:12.0f} {m.chart_analysis.profit_per_interval:12.0f} {m.max_profit:11.0f}')
		print()

def check_cxos() -> None:
	orders: typing.Sequence[ExchangeOrder] = cache.get('https://rest.fnar.net/cxos/' + config.username,
			headers={'Authorization': config.fio_api_key})
	summary: typing.Mapping[tuple[str, str], ExchangeSummary] = {
		(summary['MaterialTicker'], summary['ExchangeCode']): summary
		for summary in cache.get('https://rest.fnar.net/exchange/all')
	}
	for order in orders:
		if order['Status'] == 'FILLED':
			continue
		state = summary[order['MaterialTicker'], order['ExchangeCode']]
		if order['OrderType'] == 'BUYING' and state['Bid'] is not None and state['Bid'] > order['Limit']:
			print('outbid on', f'{order["MaterialTicker"]}.{order["ExchangeCode"]}')
		elif order['OrderType'] == 'SELLING' and state['Ask'] is not None and state['Ask'] < order['Limit']:
			print('undercut on', f'{order["MaterialTicker"]}.{order["ExchangeCode"]}')
	print()

	warehouses: typing.Sequence[Warehouse] = cache.get('https://rest.fnar.net/sites/warehouses/' + config.username,
			headers={'Authorization': config.fio_api_key})
	for warehouse in warehouses:
		if warehouse['LocationNaturalId'] in config.ignore_warehouses:
			continue
		storage: Storage = cache.get(f'https://rest.fnar.net/storage/{config.username}/{warehouse["StoreId"]}',
			headers={'Authorization': config.fio_api_key})
		if storage['WeightLoad'] > 0 or storage['VolumeLoad'] > 0:
			print('warehouse', warehouse['LocationNaturalId'], 'is not empty')
	print()

def analyze_price_chart(exchange_ticker: str, midpoint: float) -> ChartAnalysis:
	'''use price chart to estimate how long it takes to fill a bid and then an ask'''
	pcpoints: list[PriceChartPoint] = [p for p in cache.get('https://rest.fnar.net/exchange/cxpc/' + exchange_ticker)
			if p['Interval'] == 'MINUTE_FIVE']
	pcpoints.reverse()
	five_min = 5 * 60 * 1000
	asks_filled: list[AskFilled] = []
	r = ChartAnalysis(bids_filled=0, asks_filled=0, profit_per_interval=0)
	for hist in pcpoints:
		time = hist['DateEpochMs'] // five_min
		bids = asks = 0
		if hist['Low'] > midpoint:
			asks = hist['Traded']
			r.asks_filled += asks
		elif hist['High'] < midpoint:
			bids = hist['Traded']
			r.bids_filled += bids
		elif hist['High'] == hist['Low']: # all trades right at midpoint
			assert hist['High'] == midpoint
		else:
			interval_bids = (hist['High'] * hist['Traded'] - hist['Volume']) / (hist['High'] - hist['Low'])
			interval_asks = hist['Traded'] - interval_bids
			r.bids_filled += interval_bids
			r.asks_filled += interval_asks
			bids = int(interval_bids)
			asks = int(interval_asks)
		if bids and asks:
			intra_interval_trades = min(bids, asks)
			r.profit_per_interval += (hist['High'] - hist['Low'])	* intra_interval_trades
			bids -= intra_interval_trades
			asks -= intra_interval_trades
		assert bids == 0 or asks == 0
		while bids > 0 and len(asks_filled) > 0:
			profit_per_unit_per_interval = (asks_filled[-1].price - hist['Low']) / (asks_filled[-1].time - time)
			if asks_filled[-1].amount >= bids:
				r.profit_per_interval += bids * profit_per_unit_per_interval
				asks_filled[-1].amount -= bids
				if asks_filled[-1].amount == 0:
					asks_filled.pop()
				bids = 0
			else:
				r.profit_per_interval += asks_filled[-1].amount * profit_per_unit_per_interval
				bids -= asks_filled[-1].amount
				asks_filled.pop()
		if asks:
			asks_filled.append(AskFilled(price=hist['High'], amount=asks, time=time))
	return r

class ExchangeOrder(typing.TypedDict):
	MaterialTicker: str
	ExchangeCode: str
	OrderType: typing.Literal['SELLING'] | typing.Literal['BUYING']
	Status: typing.Literal['FILLED'] | typing.Literal['PARTIALLY_FILLED'] | typing.Literal['FILLED']
	Amount: int
	Limit: float

class ExchangeSummary(typing.TypedDict):
	MaterialTicker: str
	ExchangeCode: str
	Bid: float | None
	Ask: float | None

class Warehouse(typing.TypedDict):
	StoreId: str
	LocationNaturalId: str

class Storage(typing.TypedDict):
	Name: str
	StorageItems: typing.Sequence[StorageItem]
	WeightLoad: float
	VolumeLoad: float
	Type: typing.Literal['STORE'] | typing.Literal['WAREHOUSE_STORE'] | typing.Literal['FTL_FUEL_STORE'] | typing.Literal['STL_FUEL_STORE'] |  typing.Literal['SHIP_STORE']

class StorageItem(typing.TypedDict):
	MaterialTicker: str
	MaterialAmount: int

class RawPrice(typing.TypedDict):
	FullTicker: str
	MaterialTicker: str
	ExchangeCode: str
	Bid: float | None
	Ask: float | None
	HighYesterday: float | None
	LowYesterday: float | None
	AverageTraded7D: float | None # averaged daily traded volume over last 7 days

class PriceChartPoint(typing.TypedDict):
	Interval: typing.Literal['MINUTE_FIVE'] | typing.Literal['MINUTE_FIFTEEN'] | typing.Literal['MINUTE_THIRTY'] | typing.Literal['HOUR_ONE'] | typing.Literal['HOUR_TWO'] | typing.Literal['HOUR_FOUR'] | typing.Literal['HOUR_SIX'] | typing.Literal['HOUR_TWELVE'] | typing.Literal['DAY_ONE'] | typing.Literal['DAY_THREE']
	DateEpochMs: int
	High: float
	Low: float
	Volume: float
	Traded: int

@dataclasses.dataclass(eq=False, slots=True)
class AskFilled:
	price: float
	amount: int
	time: int

@dataclasses.dataclass(eq=False, slots=True)
class ChartAnalysis:
	bids_filled: float
	asks_filled: float
	profit_per_interval: float

@dataclasses.dataclass(eq=False, frozen=True, slots=True)
class Market:
	full_ticker: str
	bid: float
	ask: float
	spread: float
	max_profit: float
	chart_analysis: ChartAnalysis

	def __lt__(self, o: Market) -> bool:
		return self.chart_analysis.profit_per_interval < o.chart_analysis.profit_per_interval

if __name__ == '__main__':
	main()