pruncalc/movers.py

from __future__ import annotations

import collections
import dataclasses
import datetime
import typing

import cache

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)
	for current_price in current_prices:
		old_price = old_prices[current_price['FullTicker']]
		if (mover := analyze_raw_price(current_price, old_price)) is not None:
			movers[mover.ticker].append(mover)

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

	for commodity in top_movers:
		print(f'{commodity[0].score:9,.0f}', ' '.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=14)
	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:
		return
	if (current_vwap7d := current_price['VWAP7D']) is None or (old_vwap7d := old_price['VWAP7D']) is None:
		return
	diff = abs(current_vwap7d - old_vwap7d)
	if diff / min(current_vwap7d, old_vwap7d) > 0.15:
		return Mover(current_price['ExchangeCode'], current_price['MaterialTicker'], diff * traded)

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

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

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

if __name__ == '__main__':
	main()