pruncalc/movers.py

from __future__ import annotations

import collections
import dataclasses
import datetime
import typing

import cache

OLD_PRICE_DAYS = 14

def main() -> None:
	current_prices: typing.Sequence[RawPrice] = cache.get('https://refined-prun.github.io/refined-prices/all.json')
	old_prices = get_old_prices()

	movers: dict[str, list[Mover]] = collections.defaultdict(list)
	shortages: dict[str, list[Mover]] = collections.defaultdict(list)
	traded_map: dict[str, int] = {}
	for current_price in current_prices:
		if current_price['ExchangeCode'].endswith('2'):
			continue
		old_price = old_prices[current_price['FullTicker']]
		if (mover := analyze_raw_price(current_price, old_price)) is not None:
			if mover.score > 10000:
				movers[mover.ticker].append(mover)
			if mover.days_supply_lost > 7 and mover.days_supply_remaining < 14:
				shortages[mover.ticker].append(mover)

			assert (traded := current_price['AverageTraded30D']) is not None
			traded_map[current_price['FullTicker']] = int(traded)

	for ticker_movers in movers.values():
		ticker_movers.sort(reverse=True)
	for ticker_shortages in shortages.values():
		ticker_shortages.sort(reverse=True)
	top_movers = sorted(movers.values(), key=lambda m: m[0].score, reverse=True)
	top_shortages = sorted(shortages.values(), key=lambda m: m[0].days_supply_remaining, reverse=True)

	print('top movers:')
	for commodity in top_movers:
		print(f'{commodity[0].price_change:5,.2f}', ' '.join(f'{mover.ticker}.{mover.exchange_code}' for mover in commodity))
	
	print('\ntop shortages:')
	for commodity in top_shortages:
		print(f'{-commodity[0].days_supply_lost:6.1f}d', f'{commodity[0].supply_consumption_rate:7.1f}/d',
				' '.join(f'{mover.ticker}.{mover.exchange_code}' for mover in commodity))

	print('\nthin supply:')
	for ticker, ticker_shortages in shortages.items():
		for commodity in ticker_shortages:
			exchange_ticker = f'{commodity.ticker}.{commodity.exchange_code}'
			asks: list[Order] = cache.get(f'https://rest.fnar.net/exchange/{exchange_ticker}')['SellingOrders']
			if len(asks) == 0:
				print(f'{exchange_ticker:3}: no asks')
				continue
			asks.sort(key=lambda o: o['ItemCost'], reverse=True)
			current_ask = asks[-1]['ItemCost']
			traded = traded_map[exchange_ticker]
			remaining = traded // 2
			expected_price = float('-infinity')
			while remaining > 0:
				if len(asks) == 0:
					expected_price = float('infinity')
					break
				if asks[-1]['ItemCount'] > remaining:
					expected_price = asks[-1]['ItemCost']
					break
				remaining -= asks.pop()['ItemCount']
			if (expected_price - current_ask) / current_ask > 0.15:
				print(f'{exchange_ticker:>7}: {traded:7,} {current_ask:10,.2f} → {expected_price:10,.2f}')

def get_old_prices() -> typing.Mapping[str, RawPrice]:
	week_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=OLD_PRICE_DAYS)
	commits = cache.get(f'https://api.github.com/repos/refined-prun/refined-prices/commits?until={week_ago.isoformat()}&per_page=1')
	return {p['FullTicker']: p
			for p in cache.get(f'https://raw.githubusercontent.com/refined-prun/refined-prices/{commits[0]["sha"]}/all.json')}

def analyze_raw_price(current_price: RawPrice, old_price: RawPrice) -> Mover | None:
	if (traded := current_price['AverageTraded30D']) is None or traded < 100:
		return
	if (current_vwap7d := current_price['VWAP7D']) is None or (old_vwap7d := old_price['VWAP7D']) is None:
		return
	diff = current_vwap7d - old_vwap7d
	supply_delta = (old_price['Supply'] - current_price['Supply'])
	days_supply_lost = supply_delta / traded
	days_supply_remaining = current_price['Supply'] / traded
	if abs(diff) / min(current_vwap7d, old_vwap7d) > 0.15 or (days_supply_lost > 7 and days_supply_remaining < 14):
		return Mover(current_price['ExchangeCode'], current_price['MaterialTicker'], abs(diff) * traded,
				diff / old_vwap7d, days_supply_lost, days_supply_remaining, supply_delta / OLD_PRICE_DAYS)

class RawPrice(typing.TypedDict):
	FullTicker: str
	ExchangeCode: str
	MaterialTicker: str
	VWAP7D: float | None # volume-weighted average price over last 7 days
	AverageTraded30D: float | None # averaged daily traded volume over last 30 days
	Supply: int

class Order(typing.TypedDict):
	ItemCount: int
	ItemCost: float

@dataclasses.dataclass(eq=False, frozen=True, slots=True)
class Mover:
	exchange_code: str
	ticker: str
	score: float
	price_change: float
	days_supply_lost: float
	days_supply_remaining: float
	supply_consumption_rate: float

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

if __name__ == '__main__':
	main()