pruncalc/buy.py

from __future__ import annotations

import collections
import concurrent.futures
import dataclasses
import typing

import cache
from config import config
import supply

if typing.TYPE_CHECKING:
	import market

def main() -> None:
	with concurrent.futures.ThreadPoolExecutor() as executor:
		futures = [
			executor.submit(get_raw_prices),
			executor.submit(get_total_buy), # what we need to buy
			executor.submit(supply.warehouse_inventory), # what we have
			executor.submit(get_bids), # what we already are bidding for
		]
		raw_prices, buy, warehouse, (bids, orders) = (f.result() for f in futures)
		executor.shutdown()

	# what's left to buy
	materials: list[Material] = []
	for mat, amount in buy.items():
		remaining = amount - bids[mat] - warehouse.get(mat, 0)
		if remaining <= 0:
			continue
		price = raw_prices[mat]
		if price['Bid'] is None or price['Ask'] is None:
			print(mat, 'has no bid/ask')
			continue
		spread = price['Ask'] - price['Bid']
		materials.append(Material(mat, amount=amount, bids=bids[mat], warehouse=warehouse.get(mat, 0),
				spread=spread, savings=spread * remaining))
	materials.sort(reverse=True)

	print('mat   want  bids  have   buy  savings')
	for m in materials:
		print(f'{m.ticker:4} {m.amount:>5} {m.bids:>5} {m.warehouse:>5} {m.amount - m.bids - m.warehouse:>5} {m.savings:8.0f}')

	# deposits of current bids
	orders.sort(key=lambda order: order['Limit'] * order['Amount'], reverse=True)
	print('\ncurrent bid deposits:')
	for order in orders:
		print(f"{order['MaterialTicker']:4} {order['Limit'] * order['Amount']:7,.0f}")

def get_raw_prices() -> typing.Mapping[str, market.RawPrice]:
	return {p['MaterialTicker']: p
			for p in cache.get('https://refined-prun.github.io/refined-prices/all.json') if p['ExchangeCode'] == 'IC1'}

def get_total_buy() -> typing.Mapping[str, int]:
	planets = [supply.Planet(fio_burn) for fio_burn in cache.get('https://rest.fnar.net/fioweb/burn/user/' + config.username,
			headers={'Authorization': config.fio_api_key})]
	buy: dict[str, int] = collections.defaultdict(int)
	for planet in planets:
		for mat, amount in planet.supply_for_days(7).items():
			buy[mat] += amount
	return buy

def get_bids() -> tuple[typing.Mapping[str, int], list[market.ExchangeOrder]]:
	orders: typing.Sequence[market.ExchangeOrder] = cache.get('https://rest.fnar.net/cxos/' + config.username,
			headers={'Authorization': config.fio_api_key})
	orders = [order for order in orders
			if order['OrderType'] == 'BUYING' and order['Status'] != 'FILLED' and order['ExchangeCode'] == 'IC1']

	bids = collections.defaultdict(int)
	for order in orders:
		bids[order['MaterialTicker']] += order['Amount']
	return bids, orders

@dataclasses.dataclass(eq=False, slots=True)
class Material:
	ticker: str
	amount: int
	bids: int
	warehouse: int
	spread: float
	savings: float

	def __lt__(self, o: Material) -> bool:
		return self.savings< o.savings


if __name__ == '__main__':
	main()