pruncalc/market.py

from __future__ import annotations

import collections
import concurrent.futures
import dataclasses
import datetime
import statistics
import sys
import typing

import cache
from config import config

def main() -> None:
	raw_prices: typing.Sequence[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.profits:10.1f}')
		return

	check_warehouses()

	print()
	for health in sorted(analyze_markets(raw_prices)):
		line = f'{health.ticker:3}: {health.score:10.1f}'
		if health.my_bid >= health.lowest_bid:
			print(f'\033[90m{line}\033[0m')
		else:
			print(line)

	markets: dict[str, list[Market]] = collections.defaultdict(list)
	with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
		futures: list[concurrent.futures.Future[Market | None]] = []
		for price in raw_prices:
			futures.append(executor.submit(analyze_raw_price, price))
		for future in futures:
			if (market := future.result()) is not None:
				markets[market.exchange_code].append(market)
		executor.shutdown()

	print('\n mat       bid   ask spread  bids filled  asks filled     profit  p75 fill time')
	for commodities in markets.values():
		commodities.sort(reverse=True)
		for m in commodities:
			print(f'{m.ticker:>4}.{m.exchange_code} {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.profits:10.0f}  {format_td(m.chart_analysis.p75_fill_time)}')
		print()

def check_warehouses() -> None:
	orders: typing.Sequence[ExchangeOrder] = cache.get('https://rest.fnar.net/cxos/' + config.username,
			headers={'Authorization': config.fio_rest_key})
	trades: dict[str, list] = collections.defaultdict(list)
	for order in orders:
		if order['OrderType'] != 'BUYING' or order['Status'] == 'PLACED' or order['ExchangeCode'] != 'IC1':
			continue
		trades[order['MaterialTicker']].extend(order['Trades'])

	warehouses: typing.Sequence[Warehouse] = cache.get('https://rest.fnar.net/sites/warehouses/' + config.username,
			headers={'Authorization': config.fio_rest_key})
	now = datetime.datetime.now(datetime.UTC)
	for warehouse in warehouses:
		if warehouse['LocationNaturalId'] != 'HRT':
			continue
		storage: Storage = cache.get(f'https://rest.fnar.net/storage/{config.username}/{warehouse["StoreId"]}',
			headers={'Authorization': config.fio_rest_key})
		for item in storage['StorageItems']:
			if item['MaterialTicker'] is None:
				continue
			threshold = config.market.mm_items.get(item['MaterialTicker'])
			if threshold is not None and item['MaterialAmount'] > threshold:
				print(f'{item["MaterialAmount"] - threshold} {item["MaterialTicker"]} at {warehouse["LocationNaturalId"]}')
				mat_trades = trades.get(item['MaterialTicker'], [])
				mat_trades.sort(key=lambda t: t['TradeTimeEpochMs'], reverse=True)
				for trade in mat_trades:
					dt = datetime.datetime.fromtimestamp(trade["TradeTimeEpochMs"] // 1000, datetime.UTC)
					if now - dt > datetime.timedelta(days=7):
						break
					print(f'  {dt} {trade["PartnerName"]} {trade["Amount"]} @ {trade["Price"]}')
		break

def analyze_markets(raw_prices: typing.Sequence[RawPrice]) -> typing.Iterator[MarketHealth]:
	'''score IC1 based on how much better the other CXes are'''
	# get my top bid for every mat
	orders: typing.Sequence[ExchangeOrder] = cache.get('https://rest.fnar.net/cxos/' + config.username,
			headers={'Authorization': config.fio_rest_key})
	bids: dict[str, float] = {}
	for order in orders:
		if order['OrderType'] != 'BUYING' or order['Status'] == 'FILLED' or order['ExchangeCode'] != 'IC1':
			continue
		mat = order['MaterialTicker']
		if order['Limit'] > bids.get(mat, 0):
			bids[mat] = order['Limit']

	markets: dict[str, list[RawPrice]] = collections.defaultdict(list)
	for price in raw_prices:
		if price['ExchangeCode'].endswith('2'):
			continue
		markets[price['MaterialTicker']].append(price)

	for ticker, mat_prices in markets.items():
		(ic1_price,) = (price for price in mat_prices if price['ExchangeCode'] == 'IC1')
		highest_ask = 0
		lowest_bid = 10_000_000
		lowest_traded = 1_000_000
		for price in mat_prices:
			if price['ExchangeCode'] == 'IC1':
				continue
			if price['Ask'] is None or price['Ask'] > highest_ask:
				highest_ask = price['Ask'] or 0
			if price['Bid'] is None or price['Bid'] < lowest_bid:
				lowest_bid = price['Bid'] or 0
			if price['Traded30D'] is None or price['Traded30D'] < lowest_traded:
				lowest_traded = price['Traded30D'] or 0
		if lowest_traded == 0:
			continue
		score = 0
		if lowest_bid > 0:
			score = (lowest_bid - (ic1_price['Bid'] or 0)) / lowest_bid * 100
		if highest_ask > 0:
			score += ((ic1_price['Ask'] or 10_000_000) - highest_ask) / highest_ask * 100
		if score > 0 and (trade_activity_deficit := lowest_traded - (ic1_price['Traded30D'] or 0)) > 0:
			score *= trade_activity_deficit
		if score > 5:
			yield MarketHealth(ticker=ticker, score=score, lowest_bid=lowest_bid, my_bid=bids.get(ticker, 0))

def analyze_raw_price(price: RawPrice) -> Market | None:
	if (traded := price['AverageTraded7D']) is None or traded < 100:
		return
	if price['Bid'] is None or price['Ask'] is None:
		return
	if (high := price['HighYesterday']) is None or (low := price['LowYesterday']) is None:
		return
	if (high - low) / high < 0.1:
		return
	spread = (price['Ask'] - price['Bid']) / price['Ask']
	if (spread < 0.25 and price['ExchangeCode'] != 'IC1') or (spread < 0.15 and price['ExchangeCode'] == 'IC1'):
		return
	chart_analysis = analyze_price_chart(price['FullTicker'], (price['Bid'] + price['Ask']) / 2)
	return Market(price['ExchangeCode'], price['MaterialTicker'], bid=price['Bid'], ask=price['Ask'],
			spread=spread, chart_analysis=chart_analysis)

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
	cutoff = datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=30)
	asks_filled: list[AskFilled] = []
	fill_time = []
	r = ChartAnalysis(bids_filled=0, asks_filled=0, profits=0, p75_fill_time=datetime.timedelta.max)
	for hist in pcpoints:
		if datetime.datetime.fromtimestamp(hist['DateEpochMs'] // 1000, datetime.UTC) < cutoff:
			continue
		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.profits += (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 = (asks_filled[-1].price - hist['Low'])
			if asks_filled[-1].amount >= bids:
				r.profits += bids * profit
				asks_filled[-1].amount -= bids
				if asks_filled[-1].amount == 0:
					fill_time.append(asks_filled[-1].time - time)
					asks_filled.pop()
				bids = 0
			else:
				r.profits += asks_filled[-1].amount * profit
				bids -= asks_filled[-1].amount
				fill_time.append(asks_filled[-1].time - time)
				asks_filled.pop()
		if asks:
			asks_filled.append(AskFilled(price=hist['High'], amount=asks, time=time))
	if len(fill_time) > 0:
		r.p75_fill_time = statistics.quantiles(fill_time, n=4)[2] * datetime.timedelta(minutes=5)
	return r

def format_td(td: datetime.timedelta) -> str:
	if td == datetime.timedelta.max:
		return '∞'
	days, seconds = divmod(td.total_seconds(), 24 * 60 * 60)
	hours = seconds / (60 * 60)
	return f'{int(days)}d {hours:4.1f}h'

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

class ExchangeTrade(typing.TypedDict):
	TradeTimeEpochMs: int
	Amount: int
	Price: float
	PartnerName: str

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', 'WAREHOUSE_STORE', 'FTL_FUEL_STORE', 'STL_FUEL_STORE', 'SHIP_STORE']

class StorageItem(typing.TypedDict):
	MaterialTicker: str | None # shipment blocks are None
	MaterialAmount: int
	Type: typing.Literal['INVENTORY', 'SHIPMENT']

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
	Traded7D: int | None
	Traded30D: int | None

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

@dataclasses.dataclass(eq=False, slots=True)
class MarketHealth:
	ticker: str
	score: float
	lowest_bid: float
	my_bid: float

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

@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
	profits: float
	p75_fill_time: datetime.timedelta

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

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

if __name__ == '__main__':
	main()