from __future__ import annotations import dataclasses import typing import cache from config import config def main() -> None: check_cxos() raw_prices: list[RawPrice] = cache.get('https://refined-prun.github.io/refined-prices/all.json') markets: list[Market] = [] for price in raw_prices: if (traded := price['AverageTraded7D']) is None or traded < 100: continue if price['Bid'] is None or price['Ask'] is None: continue if (high := price['HighYesterday']) is None or (low := price['LowYesterday']) is None: continue if (high - low) / high < 0.1: continue spread = (price['Ask'] - price['Bid']) / price['Ask'] if spread < 0.15: continue markets.append(Market(price['FullTicker'], bid=price['Bid'], ask=price['Ask'], spread=spread, traded=traded)) markets.sort(key=lambda m: (m.ask - m.bid) * m.traded, reverse=True) print(f'{"mat":^8} {"bid":^5} {"ask":^5} spread {"traded":^7} ') for market in markets: print(f'{market.full_ticker:>8} {market.bid:5} {market.ask:5} {market.spread*100: 5.0f}% {market.traded:7}') def check_cxos() -> None: orders: typing.Sequence[ExchangeOrder] = cache.get('https://rest.fnar.net/cxos/' + config.username, headers={'Authorization': config.fio_api_key}) summary: typing.Mapping[tuple[str, str], ExchangeSummary] = { (summary['MaterialTicker'], summary['ExchangeCode']): summary for summary in cache.get('https://rest.fnar.net/exchange/all') } for order in orders: state = summary[order['MaterialTicker'], order['ExchangeCode']] if order['OrderType'] == 'BUYING' and state['Bid'] is not None and state['Bid'] > order['Limit']: print('outbid on', f'{order["MaterialTicker"]}.{order["ExchangeCode"]}') elif order['OrderType'] == 'SELLING' and state['Ask'] is not None and state['Ask'] < order['Limit']: print('undercut on', f'{order["MaterialTicker"]}.{order["ExchangeCode"]}') print() warehouses: typing.Sequence[Warehouse] = cache.get('https://rest.fnar.net/sites/warehouses/' + config.username, headers={'Authorization': config.fio_api_key}) for warehouse in warehouses: if warehouse['LocationNaturalId'] in config.ignore_warehouses: continue storage: Storage = cache.get(f'https://rest.fnar.net/storage/{config.username}/{warehouse["StoreId"]}', headers={'Authorization': config.fio_api_key}) if storage['WeightLoad'] > 0 or storage['VolumeLoad'] > 0: print('warehouse', warehouse['LocationNaturalId'], 'is not empty') print() class ExchangeOrder(typing.TypedDict): MaterialTicker: str ExchangeCode: str OrderType: typing.Literal['SELLING'] | typing.Literal['BUYING'] Limit: float class ExchangeSummary(typing.TypedDict): MaterialTicker: str ExchangeCode: str Bid: float | None Ask: float | None class Warehouse(typing.TypedDict): StoreId: str LocationNaturalId: str class Storage(typing.TypedDict): StorageItems: typing.Sequence WeightLoad: float VolumeLoad: float class RawPrice(typing.TypedDict): FullTicker: str Bid: float | None Ask: float | None HighYesterday: float | None LowYesterday: float | None AverageTraded7D: float | None # averaged daily traded volume over last 7 days @dataclasses.dataclass(eq=False, slots=True) class Market: full_ticker: str bid: float ask: float spread: float traded: float if __name__ == '__main__': main()