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(max_workers=2) 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
	bid_deposits: dict[str, float] = collections.defaultdict(float)
	for order in orders:
		bid_deposits[order['MaterialTicker']] += order['Limit'] * order['Amount']
	print('\ncurrent bid deposits:')
	for mat, deposit in sorted(bid_deposits.items(), key=lambda kv: kv[1], reverse=True):
		print(f"{mat:4} {deposit: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})]

	total_consumption: dict[str, float] = collections.defaultdict(float)
	for planet in planets:
		supply_config = config.supply_config(planet.name)
		for mat, amount in planet.net_consumption.items():
			if mat not in supply_config.ignore_materials:
				total_consumption[mat] += amount

	buy: dict[str, int] = collections.defaultdict(int)
	for mat, consumption in total_consumption.items():
		if consumption <= 0:
			continue
		buy[mat] = round(consumption * days)
	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()