|
|
@@ -22,11 +22,15 @@ def main() -> None:
|
|
|
print(f'{ticker}: bids filled = {a.bids_filled:6.0f}, asks filled = {a.asks_filled:6.0f}, profit per interval = {a.profits:10.1f}')
|
|
|
return
|
|
|
|
|
|
- check_cxos()
|
|
|
+ check_warehouses()
|
|
|
|
|
|
print()
|
|
|
- for score, ticker in sorted(analyze_markets(raw_prices)):
|
|
|
- print(f'{ticker:3}: {score:10.1f}')
|
|
|
+ for health in sorted(analyze_markets(raw_prices)):
|
|
|
+ line = f'{health.ticker:3}: {health.score:10.1f}'
|
|
|
+ if health.my_bid >= health.lowest_bid:
|
|
|
+ print(f'\033[90m{line}\033[0m')
|
|
|
+ else:
|
|
|
+ print(line)
|
|
|
|
|
|
markets: dict[str, list[Market]] = collections.defaultdict(list)
|
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
|
|
|
@@ -46,7 +50,7 @@ def main() -> None:
|
|
|
f'{m.chart_analysis.asks_filled:12.0f} {m.chart_analysis.profits:10.0f} {format_td(m.chart_analysis.p75_fill_time)}')
|
|
|
print()
|
|
|
|
|
|
-def check_cxos() -> None:
|
|
|
+def check_warehouses() -> None:
|
|
|
warehouses: typing.Sequence[Warehouse] = cache.get('https://rest.fnar.net/sites/warehouses/' + config.username,
|
|
|
headers={'Authorization': config.fio_rest_key})
|
|
|
for warehouse in warehouses:
|
|
|
@@ -59,8 +63,19 @@ 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]]:
|
|
|
+def analyze_markets(raw_prices: typing.Sequence[RawPrice]) -> typing.Iterator[MarketHealth]:
|
|
|
'''score IC1 based on how much better the other CXes are'''
|
|
|
+ # get my top bid for every mat
|
|
|
+ orders: typing.Sequence[ExchangeOrder] = cache.get('https://rest.fnar.net/cxos/' + config.username,
|
|
|
+ headers={'Authorization': config.fio_rest_key})
|
|
|
+ bids: dict[str, float] = {}
|
|
|
+ for order in orders:
|
|
|
+ if order['OrderType'] != 'BUYING' or order['Status'] == 'FILLED' or order['ExchangeCode'] != 'IC1':
|
|
|
+ continue
|
|
|
+ mat = order['MaterialTicker']
|
|
|
+ if order['Limit'] > bids.get(mat, 0):
|
|
|
+ bids[mat] = order['Limit']
|
|
|
+
|
|
|
markets: dict[str, list[RawPrice]] = collections.defaultdict(list)
|
|
|
for price in raw_prices:
|
|
|
if price['ExchangeCode'].endswith('2'):
|
|
|
@@ -89,7 +104,7 @@ def analyze_markets(raw_prices: typing.Sequence[RawPrice]) -> typing.Iterator[tu
|
|
|
if score > 0 and (trade_activity_deficit := lowest_traded - (ic1_price['Traded30D'] or 0)) > 0:
|
|
|
score *= trade_activity_deficit
|
|
|
if score > 5:
|
|
|
- yield score, ticker
|
|
|
+ yield MarketHealth(ticker=ticker, score=score, lowest_bid=lowest_bid, my_bid=bids.get(ticker, 0))
|
|
|
|
|
|
def analyze_raw_price(price: RawPrice) -> Market | None:
|
|
|
if (traded := price['AverageTraded7D']) is None or traded < 100:
|
|
|
@@ -219,6 +234,16 @@ class PriceChartPoint(typing.TypedDict):
|
|
|
Volume: float
|
|
|
Traded: int
|
|
|
|
|
|
+@dataclasses.dataclass(eq=False, slots=True)
|
|
|
+class MarketHealth:
|
|
|
+ ticker: str
|
|
|
+ score: float
|
|
|
+ lowest_bid: float
|
|
|
+ my_bid: float
|
|
|
+
|
|
|
+ def __lt__(self, o: MarketHealth) -> bool:
|
|
|
+ return self.score < o.score
|
|
|
+
|
|
|
@dataclasses.dataclass(eq=False, slots=True)
|
|
|
class AskFilled:
|
|
|
price: float
|