market.py 3.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798
  1. from __future__ import annotations
  2. import dataclasses
  3. import typing
  4. import cache
  5. from config import config
  6. def main() -> None:
  7. check_cxos()
  8. raw_prices: list[RawPrice] = cache.get('https://refined-prun.github.io/refined-prices/all.json')
  9. markets: list[Market] = []
  10. for price in raw_prices:
  11. if (traded := price['AverageTraded7D']) is None or traded < 100:
  12. continue
  13. if price['Bid'] is None or price['Ask'] is None:
  14. continue
  15. if (high := price['HighYesterday']) is None or (low := price['LowYesterday']) is None:
  16. continue
  17. if (high - low) / high < 0.1:
  18. continue
  19. spread = (price['Ask'] - price['Bid']) / price['Ask']
  20. if spread < 0.15:
  21. continue
  22. markets.append(Market(price['FullTicker'], bid=price['Bid'], ask=price['Ask'], spread=spread, traded=traded))
  23. markets.sort(key=lambda m: (m.ask - m.bid) * m.traded, reverse=True)
  24. print(f'{"mat":^8} {"bid":^5} {"ask":^5} spread {"traded":^7} ')
  25. for market in markets:
  26. print(f'{market.full_ticker:>8} {market.bid:5} {market.ask:5} {market.spread*100: 5.0f}% {market.traded:7}')
  27. def check_cxos() -> None:
  28. orders: typing.Sequence[ExchangeOrder] = cache.get('https://rest.fnar.net/cxos/' + config.username,
  29. headers={'Authorization': config.fio_api_key})
  30. summary: typing.Mapping[tuple[str, str], ExchangeSummary] = {
  31. (summary['MaterialTicker'], summary['ExchangeCode']): summary
  32. for summary in cache.get('https://rest.fnar.net/exchange/all')
  33. }
  34. for order in orders:
  35. state = summary[order['MaterialTicker'], order['ExchangeCode']]
  36. if order['OrderType'] == 'BUYING' and state['Bid'] is not None and state['Bid'] > order['Limit']:
  37. print('outbid on', f'{order["MaterialTicker"]}.{order["ExchangeCode"]}')
  38. elif order['OrderType'] == 'SELLING' and state['Ask'] is not None and state['Ask'] < order['Limit']:
  39. print('undercut on', f'{order["MaterialTicker"]}.{order["ExchangeCode"]}')
  40. print()
  41. warehouses: typing.Sequence[Warehouse] = cache.get('https://rest.fnar.net/sites/warehouses/' + config.username,
  42. headers={'Authorization': config.fio_api_key})
  43. for warehouse in warehouses:
  44. if warehouse['LocationNaturalId'] in config.ignore_warehouses:
  45. continue
  46. storage: Storage = cache.get(f'https://rest.fnar.net/storage/{config.username}/{warehouse["StoreId"]}',
  47. headers={'Authorization': config.fio_api_key})
  48. if storage['WeightLoad'] > 0 or storage['VolumeLoad'] > 0:
  49. print('warehouse', warehouse['LocationNaturalId'], 'is not empty')
  50. print()
  51. class ExchangeOrder(typing.TypedDict):
  52. MaterialTicker: str
  53. ExchangeCode: str
  54. OrderType: typing.Literal['SELLING'] | typing.Literal['BUYING']
  55. Limit: float
  56. class ExchangeSummary(typing.TypedDict):
  57. MaterialTicker: str
  58. ExchangeCode: str
  59. Bid: float | None
  60. Ask: float | None
  61. class Warehouse(typing.TypedDict):
  62. StoreId: str
  63. LocationNaturalId: str
  64. class Storage(typing.TypedDict):
  65. StorageItems: typing.Sequence
  66. WeightLoad: float
  67. VolumeLoad: float
  68. class RawPrice(typing.TypedDict):
  69. FullTicker: str
  70. Bid: float | None
  71. Ask: float | None
  72. HighYesterday: float | None
  73. LowYesterday: float | None
  74. AverageTraded7D: float | None # averaged daily traded volume over last 7 days
  75. @dataclasses.dataclass(eq=False, slots=True)
  76. class Market:
  77. full_ticker: str
  78. bid: float
  79. ask: float
  80. spread: float
  81. traded: float
  82. if __name__ == '__main__':
  83. main()