market: find mats to MM

market.py

@@ -12,7 +12,7 @@ import cache
 from config import config
 
 def main() -> None:
-	raw_prices: list[RawPrice] = cache.get('https://refined-prun.github.io/refined-prices/all.json')
+	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:]
@@ -24,6 +24,10 @@ def main() -> None:
 
 	check_cxos()
 
+	print()
+	for score, ticker in sorted(analyze_markets(raw_prices)):
+		print(ticker, score)
+
 	markets: dict[str, list[Market]] = collections.defaultdict(list)
 	with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
 		futures: list[concurrent.futures.Future[Market | None]] = []
@@ -55,6 +59,37 @@ def check_cxos() -> None:
 			if threshold is not None and item['MaterialAmount'] > threshold:
 				print(f'{item["MaterialAmount"] - threshold} {item["MaterialTicker"]} at {warehouse["LocationNaturalId"]}')
 
+def analyze_markets(raw_prices: typing.Sequence[RawPrice]) -> typing.Iterator[tuple[float, str]]:
+	'''score IC1 based on how much better the other CXes are'''
+	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 10_000_000
+			if price['Traded30D'] is None or price['Traded30D'] < lowest_traded:
+				lowest_traded = price['Traded30D'] or 0
+		if lowest_traded == 0:
+			continue
+		score = (ic1_price['Ask'] or 10_000_000) - highest_ask
+		score += lowest_bid - (ic1_price['Bid'] or 0)
+		if score > 0 and (trade_activity_deficit := lowest_traded - (ic1_price['Traded30D'] or 0)) > 0:
+			score *= trade_activity_deficit
+		if score > 500:
+			yield score, ticker
+
 def analyze_raw_price(price: RawPrice) -> Market | None:
 	if (traded := price['AverageTraded7D']) is None or traded < 100:
 		return
@@ -65,7 +100,7 @@ def analyze_raw_price(price: RawPrice) -> Market | None:
 	if (high - low) / high < 0.1:
 		return
 	spread = (price['Ask'] - price['Bid']) / price['Ask']
-	if spread < 0.15:
+	if spread < 0.25:
 		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'],
@@ -173,6 +208,7 @@ class RawPrice(typing.TypedDict):
 	HighYesterday: float | None
 	LowYesterday: float | None
 	AverageTraded7D: float | None # averaged daily traded volume over last 7 days
+	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']