| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798 |
- 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()
|