pruncalc/buy.py

from __future__ import annotations

import collections
import concurrent.futures
import dataclasses
import json
import math
import sys
import typing

import cache
from config import config
import supply

if typing.TYPE_CHECKING:
	import market

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

	# what's left to buy
	materials: list[Material] = []
	price_limits: dict[str, int] = {}
	for mat, amount in buy.items():
		remaining = max(amount - bids[mat] - warehouse.get(mat, 0) - exports.get(mat, 0), 0)
		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], have=warehouse.get(mat, 0) + exports.get(mat, 0),
				spread=spread, savings=spread * remaining))
		epsilon = 10 ** (int(math.log10(price['Bid'])) - 2)
		price_limits[mat] = price['Bid'] + 2 * epsilon
	materials.sort(reverse=True)

	to_buy: dict[str, int] = {}
	print('mat   want  bids  have   buy  savings')
	for m in materials:
		buy = max(m.amount - m.bids - m.have, 0)
		if m.bids == 0 and buy > 0:
			bids = f'\033[91m{m.bids:5}\033[0m'
		else:
			bids = str(m.bids).rjust(5)
		print(f'{m.ticker:4} {m.amount:>5} {bids} {m.have:>5} {buy:>5} {m.savings:8.0f}')
		if buy > 0:
			to_buy[m.ticker] = buy

	print('\n' + json.dumps({
		'actions': [
			{'name': 'BuyItems', 'type': 'CX Buy', 'group': 'A1', 'exchange': 'IC1',
				'priceLimits': price_limits, 'buyPartial': True, 'allowUnfilled': True, 'useCXInv': False},
		],
		'global': {'name': f'buy orders for {days} days'},
		'groups': [{'type': 'Manual', 'name': 'A1', 'materials': to_buy}],
	}))

	# 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(days: int) -> 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(days).items():
			buy[mat] += amount
	return buy

def get_planet_exports_and_ship_storage() -> typing.Mapping[str, int]:
	'''materials in base storage that aren't being consumed and materials in ship storage'''
	avail = collections.defaultdict(int)

	fio_burn: typing.Sequence[supply.FIOBurn] = cache.get('https://rest.fnar.net/fioweb/burn/user/' + config.username,
			headers={'Authorization': config.fio_api_key})
	for burn in fio_burn:
		planet = supply.Planet(burn)
		for mat in planet.exporting:
			avail[mat] += planet.inventory.get(mat, 0)
	
	stores: typing.Sequence[market.Storage] = cache.get('https://rest.fnar.net/storage/' + config.username,
			headers={'Authorization': config.fio_api_key})
	for store in stores:
		if store['Type'] != 'SHIP_STORE':
			continue
		for item in store['StorageItems']:
			avail[item['MaterialTicker']] += item['MaterialAmount']

	return avail

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, frozen=True, slots=True)
class Material:
	ticker: str
	amount: int
	bids: int
	have: int
	spread: float
	savings: float

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

if __name__ == '__main__':
	main()