pruncalc/market.py

from __future__ import annotations

import collections
import dataclasses
import sys
import typing

import cache
from config import config

def main() -> None:
	if len(sys.argv) > 1:
		exchange_tickers = sys.argv[1:]
		for ticker in exchange_tickers:
			bids_filled, asks_filled = estimate_filled_orders(ticker, 0)
			print(f'{ticker}: {bids_filled=}, {asks_filled=}')
		return

	check_cxos()

	raw_prices: list[RawPrice] = cache.get('https://refined-prun.github.io/refined-prices/all.json')
	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
		bids_filled, asks_filled = estimate_filled_orders(price['FullTicker'], (price['Bid'] + price['Ask']) / 2)
		bid_fill_ratio = bids_filled / (bids_filled + asks_filled)
		if bid_fill_ratio < 0.05 or bid_fill_ratio > 0.95:
			continue
		max_profit = (price['Ask'] - price['Bid']) * min(bids_filled, asks_filled)
		markets[price['ExchangeCode']].append(Market(price['FullTicker'], bid=price['Bid'], ask=price['Ask'],
				spread=spread, traded=traded, bids_filled=bids_filled, asks_filled=asks_filled, max_profit=max_profit))

	print(f'{"mat":^8} {"bid":^5} {"ask":^5} spread  {"traded":^7}  bids filled  asks filled  max profit')
	for commodities in markets.values():
		for m in commodities:
			print(f'{m.full_ticker:>8} {m.bid:5} {m.ask:5} {m.spread*100: 5.0f}% {m.traded:7.0f} {m.bids_filled:10.0f}   {m.asks_filled:10.0f}   {m.max_profit:10.0f}')

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:
		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 estimate_filled_orders(exchange_ticker: str, midpoint: float) -> tuple[float, float]:
	'''use price chart to estimate how many bids and asks were filled'''
	bids = asks = 0
	for hist in cache.get('https://rest.fnar.net/exchange/cxpc/' + exchange_ticker):
		if hist['Interval'] != 'MINUTE_FIVE':
			continue
		if hist['Low'] > midpoint:
			asks += hist['Traded']
		elif hist['High'] < midpoint:
			bids += hist['Traded']
		elif hist['High'] == hist['Low']:
			assert hist['High'] == midpoint
		else:
			interval_bids = (hist['High'] * hist['Traded'] - hist['Volume']) / (hist['High'] - hist['Low'])
			interval_asks = hist['Traded'] - interval_bids
			bids += interval_bids
			asks += interval_asks
	return bids, asks

class ExchangeOrder(typing.TypedDict):
	MaterialTicker: str
	ExchangeCode: str
	OrderType: typing.Literal['SELLING'] | typing.Literal['BUYING']
	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):
	StorageItems: typing.Sequence
	WeightLoad: float
	VolumeLoad: float

class RawPrice(typing.TypedDict):
	FullTicker: 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']
	High: float
	Low: float
	Volume: float
	Traded: int

@dataclasses.dataclass(eq=False, slots=True)
class Market:
	full_ticker: str
	bid: float
	ask: float
	spread: float
	traded: float
	bids_filled: float
	asks_filled: float
	max_profit: float

	def __lt__(self, o) -> bool:
		return self.max_profit < o.max_profit

if __name__ == '__main__':
	main()