movers: analyze shortages

movers.py

@@ -7,38 +7,58 @@ 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:
-			movers[mover.ticker].append(mover)
+			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=14)
+	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:
+	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
-	if abs(diff) / min(current_vwap7d, old_vwap7d) > 0.15:
+	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)
+				diff / old_vwap7d, days_supply_lost, days_supply_remaining, supply_delta / OLD_PRICE_DAYS)
 
 class RawPrice(typing.TypedDict):
 	FullTicker: str
@@ -46,6 +66,7 @@ class RawPrice(typing.TypedDict):
 	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:
@@ -53,6 +74,9 @@ class Mover:
 	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