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)
	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)

	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))

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

@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()