diff options
Diffstat (limited to 'backend/src/manager.py')
| -rw-r--r-- | backend/src/manager.py | 483 |
1 files changed, 454 insertions, 29 deletions
diff --git a/backend/src/manager.py b/backend/src/manager.py index 09d920e..7e86949 100644 --- a/backend/src/manager.py +++ b/backend/src/manager.py @@ -1,38 +1,88 @@ -import threading +import asyncio +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from typing import Any +from time import perf_counter from graph import NavigationGraph -from esi.api import get_location, set_waypoints, open_market_window +from esi.api import WalletTransaction, get_location, get_wallet_transactions, open_market_window, set_waypoints as push_waypoints from market import Market -from arbitrage import arbitrage, snipe -from pathfinder import pathfinder +from arbitrage import ShortlistConfig, Trade, arbitrage, snipe +from pathfinder import calculate_net_profit, filter_order_matches_low_profit, filter_order_matches_not_in_graph import route +PRIMARY_SHORTLIST = ShortlistConfig() +FALLBACK_SHORTLIST = ShortlistConfig(candidate_limit=80_000, type_limit=600, station_pair_limit=80) +WALLET_POLL_INTERVAL_SECONDS = 60.0 +TRADE_CONFIRMATION_TTL = timedelta(hours=6) +PRICE_TOLERANCE_RATIO = 0.002 +MIN_PRICE_TOLERANCE = 0.01 +CLOCK_SKEW_TOLERANCE = timedelta(minutes=5) + + +@dataclass +class TrackedTradeConfirmation: + trade: Trade + created_at: datetime + expires_at: datetime + buy_remaining: int + sell_remaining: int + + @classmethod + def from_trade(cls, trade: Trade, created_at: datetime) -> "TrackedTradeConfirmation": + return cls( + trade=trade, + created_at=created_at, + expires_at=created_at + TRADE_CONFIRMATION_TTL, + buy_remaining=trade.quantity, + sell_remaining=trade.quantity, + ) + + def is_complete(self) -> bool: + return self.buy_remaining <= 0 and self.sell_remaining <= 0 + + def is_expired(self, now: datetime) -> bool: + return now >= self.expires_at + + class Manager: def __init__(self): self.ship = { - "location": get_location(), + "location": None, "cargo": 4000, "max_warp_speed": 4.55, # AU/s "max_subwarp_speed": 166.9, # m/s "gankers_areas": {"Ahbazon": 0.5}, #["Miroitem", "Rancer", "Huola", "Ahbazon", "Uedama", "Sivala", "Tama", "Onatoh"], "player_cost": 15_000_000 / 3600, # isk/second "risk_cost": 60_000_000, # isk - "ship_cost": 5_500_000 # isk + "ship_cost": 5_500_000, # isk + "tax_rate": 8.0, # percentage % + "profit_threshold": 100_000, + "max_capital": 20_000_000, + "snipe_profit_threshold": 20_000_000 } self.graph = NavigationGraph(self.ship) self.route = [] self.route_info = {} + self.route_path: list[int] = [] + self.route_path_offsets: list[int] = [] + self.active_route_trades: list[Trade] = [] + self.active_trade_confirmations: list[TrackedTradeConfirmation] = [] + self.pending_trade_confirmations: list[TrackedTradeConfirmation] = [] self.market = Market() self.updating_orders = False self.log = ["Test Log"] + self.last_wallet_poll_at = 0.0 + self.last_seen_wallet_transaction_id = 0 + self.wallet_scope_error_logged = False - def route(self): + def get_route(self): return self.route - def route_info(self): + def get_route_info(self): return self.route_info def get_log(self): @@ -40,7 +90,349 @@ class Manager: self.log = [] return result - def snipe_callback(self, updated_regions): + def build_trade_confirmations(self, trades: list[Trade]) -> list[TrackedTradeConfirmation]: + created_at = datetime.now(UTC) + return [TrackedTradeConfirmation.from_trade(trade, created_at) for trade in trades] + + def clear_active_route(self) -> None: + self.route = [] + self.route_info = {} + self.route_path = [] + self.route_path_offsets = [] + self.active_route_trades = [] + self.active_trade_confirmations = [] + + def trim_active_route(self, display_start_index: int) -> None: + if display_start_index <= 0: + return + + self.route = self.route[display_start_index:] + + if display_start_index < len(self.route_path_offsets): + raw_start_index = self.route_path_offsets[display_start_index] + self.route_path = self.route_path[raw_start_index:] + self.route_path_offsets = [ + offset - raw_start_index + for offset in self.route_path_offsets[display_start_index:] + ] + else: + self.route_path = [] + self.route_path_offsets = [] + + def refresh_route_info(self) -> None: + if not self.route or not self.active_route_trades or not self.route_path: + self.route_info = {} + return + + self.route_info = route.calculate_route_info( + self.route, + self.active_route_trades, + self.graph, + self.ship, + metric_path=self.route_path, + ) or {} + + def prune_trade_confirmations(self) -> None: + now = datetime.now(UTC) + self.active_trade_confirmations = [ + confirmation + for confirmation in self.active_trade_confirmations + if not confirmation.is_complete() + ] + + retained_pending: list[TrackedTradeConfirmation] = [] + expired_confirmations = 0 + for confirmation in self.pending_trade_confirmations: + if confirmation.is_complete(): + continue + if confirmation.is_expired(now): + expired_confirmations += 1 + continue + retained_pending.append(confirmation) + + self.pending_trade_confirmations = retained_pending + if expired_confirmations: + self.log.append(f"Dropped {expired_confirmations} stale unconfirmed trade confirmations") + + def move_active_confirmations_to_pending(self) -> None: + outstanding_confirmations = [ + confirmation + for confirmation in self.active_trade_confirmations + if not confirmation.is_complete() + ] + if not outstanding_confirmations: + self.active_trade_confirmations = [] + return + + now = datetime.now(UTC) + for confirmation in outstanding_confirmations: + confirmation.expires_at = max(confirmation.expires_at, now + TRADE_CONFIRMATION_TTL) + + self.pending_trade_confirmations.extend(outstanding_confirmations) + self.active_trade_confirmations = [] + self.log.append(f"Waiting for ESI confirmation on {len(outstanding_confirmations)} route trade(s)") + + def wallet_transaction_matches( + self, + transaction: WalletTransaction, + confirmation: TrackedTradeConfirmation, + ) -> bool: + trade = confirmation.trade + expected_location = trade.from_station if transaction.is_buy else trade.to_station + expected_price = trade.from_price if transaction.is_buy else trade.to_price + remaining_quantity = confirmation.buy_remaining if transaction.is_buy else confirmation.sell_remaining + + if remaining_quantity <= 0: + return False + if transaction.type_id != trade.type_id or transaction.location_id != expected_location: + return False + if transaction.date < confirmation.created_at - CLOCK_SKEW_TOLERANCE: + return False + if transaction.date > confirmation.expires_at: + return False + + price_tolerance = max(MIN_PRICE_TOLERANCE, expected_price * PRICE_TOLERANCE_RATIO) + return abs(transaction.unit_price - expected_price) <= price_tolerance + + def apply_wallet_transaction( + self, + transaction: WalletTransaction, + confirmations: list[TrackedTradeConfirmation], + order_adjustments: dict[int, int], + ) -> int: + remaining_quantity = transaction.quantity + + for confirmation in confirmations: + if not self.wallet_transaction_matches(transaction, confirmation): + continue + + remaining_attr = "buy_remaining" if transaction.is_buy else "sell_remaining" + order_id = confirmation.trade.from_order_id if transaction.is_buy else confirmation.trade.to_order_id + available_quantity = getattr(confirmation, remaining_attr) + applied_quantity = min(available_quantity, remaining_quantity) + if applied_quantity <= 0: + continue + + setattr(confirmation, remaining_attr, available_quantity - applied_quantity) + order_adjustments[order_id] = order_adjustments.get(order_id, 0) + applied_quantity + remaining_quantity -= applied_quantity + if remaining_quantity == 0: + break + + return transaction.quantity - remaining_quantity + + async def reconcile_trade_confirmations(self) -> None: + self.prune_trade_confirmations() + if not (self.active_trade_confirmations or self.pending_trade_confirmations): + return + + if perf_counter() - self.last_wallet_poll_at < WALLET_POLL_INTERVAL_SECONDS: + return + + self.last_wallet_poll_at = perf_counter() + + try: + transactions = await get_wallet_transactions() + self.wallet_scope_error_logged = False + except PermissionError as exc: + if not self.wallet_scope_error_logged: + self.log.append(str(exc)) + self.wallet_scope_error_logged = True + return + except Exception as exc: + self.log.append(f"Failed to poll wallet transactions: {exc}") + return + + if not transactions: + return + + highest_transaction_id = max(transaction.transaction_id for transaction in transactions) + new_transactions = [ + transaction + for transaction in transactions + if transaction.transaction_id > self.last_seen_wallet_transaction_id + ] + self.last_seen_wallet_transaction_id = max(self.last_seen_wallet_transaction_id, highest_transaction_id) + + if not new_transactions: + return + + order_adjustments: dict[int, int] = {} + matched_transactions = 0 + matched_units = 0 + confirmations = self.active_trade_confirmations + self.pending_trade_confirmations + + for transaction in sorted(new_transactions, key=lambda item: (item.date, item.transaction_id)): + applied_quantity = self.apply_wallet_transaction(transaction, confirmations, order_adjustments) + if applied_quantity > 0: + matched_transactions += 1 + matched_units += applied_quantity + + self.prune_trade_confirmations() + + if not order_adjustments: + return + + summary = await self.market.apply_order_volume_adjustments(order_adjustments) + self.log.append( + f"Confirmed {matched_units:,} routed units across {matched_transactions:,} wallet transaction(s); " + f"updated {summary['affected_orders']:,} local market orders" + ) + if summary["ignored_orders"]: + self.log.append( + f"Skipped {summary['ignored_orders']:,} confirmed order adjustments because newer market data was already present" + ) + + def extract_route_waypoints(self) -> tuple[list[int], str | None]: + if not self.route: + return [], "No route exists" + + if len(self.route) <= 1: + return [], "No destination waypoints available" + + waypoints = [] + invalid_entries = [] + for step in self.route[1:]: + location_id = step.get("location_id") if isinstance(step, dict) else None + if isinstance(location_id, int): + waypoints.append(location_id) + else: + invalid_entries.append(step) + + if invalid_entries: + return [], "Current route contains invalid waypoint entries" + + if not waypoints: + return [], "No destination waypoints available" + + return waypoints, None + + async def set_waypoints(self) -> dict[str, Any]: + waypoints, error_message = self.extract_route_waypoints() + if error_message: + self.log.append(f"Waypoint push skipped: {error_message}") + return { + "success": False, + "status": "error", + "message": error_message, + "requested_count": 0, + "success_count": 0, + "applied_waypoints": [], + "failed_waypoints": [], + "errors": [], + } + + self.log.append(f"Submitting {len(waypoints)} waypoint(s) to EVE...") + result = await push_waypoints(waypoints) + requested_count = result.get("requested_count", len(waypoints)) + success_count = result.get("success_count", 0) + failed_waypoints = result.get("failed_waypoints", []) + + if result.get("success"): + message = f"Set {success_count} waypoint(s) in game" + status = "success" + elif success_count > 0: + message = f"Set {success_count}/{requested_count} waypoint(s) in game" + status = "partial" + else: + message = "Failed to set waypoints in game" + status = "error" + + self.log.append(message) + if failed_waypoints: + self.log.append(f"Failed waypoint IDs: {', '.join(str(waypoint) for waypoint in failed_waypoints)}") + if result.get("errors"): + self.log.append(f"Waypoint error: {result['errors'][0]}") + + return { + **result, + "status": status, + "message": message, + } + + def summarize_trades(self, trades): + if not trades: + return { + "count": 0, + "sources": 0, + "pairs": 0, + "triplets": 0, + } + + return { + "count": len(trades), + "sources": len({trade.from_station for trade in trades}), + "pairs": len({(trade.from_station, trade.to_station) for trade in trades}), + "triplets": len({(self.ship["location"], trade.from_station, trade.to_station) for trade in trades}), + } + + def describe_trade(self, trade): + if not trade: + return "No scored candidates" + + from_station = self.graph.stations.get(trade.from_station, {}).get("item_name", trade.from_station) + to_station = self.graph.stations.get(trade.to_station, {}).get("item_name", trade.to_station) + net_profit = trade.net_profit if trade.net_profit is not None else 0 + return ( + f"{trade.item_name} {from_station} -> {to_station} " + f"gross {trade.gross_profit:,.0f} ISK, net {net_profit:,.0f} ISK" + ) + + async def run_route_pass(self, label, shortlist): + self.log.append(f"{label}: fetching market data and shortlisting arbitrage opportunities...") + stage_start = perf_counter() + order_matches = await arbitrage( + self.market, + self.ship["cargo"], + self.ship["tax_rate"] / 100.0, + self.ship["profit_threshold"], + self.ship["max_capital"], + shortlist=shortlist, + ) + shortlist_duration = perf_counter() - stage_start + shortlist_summary = self.summarize_trades(order_matches) + self.log.append( + f"{label}: shortlisted {shortlist_summary['count']:,} candidates across {shortlist_summary['pairs']:,} station pairs " + f"and {shortlist_summary['sources']:,} source stations in {shortlist_duration:.2f}s" + ) + + self.log.append(f"{label}: preparing graph leg metrics and estimating net profit...") + stage_start = perf_counter() + graph_valid_matches = filter_order_matches_not_in_graph(order_matches, self.graph) + graph_valid_summary = self.summarize_trades(graph_valid_matches) + graph_rejects = len(order_matches) - len(graph_valid_matches) + leg_metrics = self.graph.build_trade_leg_metrics(self.ship["location"], graph_valid_matches) + scored_matches = calculate_net_profit(graph_valid_matches, self.ship["location"], leg_metrics) + path_missing = sum(1 for trade in scored_matches if trade.net_profit is None) + profitable_matches = filter_order_matches_low_profit(scored_matches) + positive_count = len(profitable_matches) + scored_count = len(scored_matches) - path_missing + non_positive = max(scored_count - positive_count, 0) + profitable_duration = perf_counter() - stage_start + profitable_summary = self.summarize_trades(profitable_matches) + best_candidate = max( + (trade for trade in scored_matches if trade.net_profit is not None), + key=lambda trade: trade.net_profit, + default=None, + ) + self.log.append( + f"{label}: retained {profitable_summary['count']:,} profitable candidates across {profitable_summary['triplets']:,} unique triplets " + f"from {graph_valid_summary['pairs']:,} graph-valid station pairs in {profitable_duration:.2f}s" + ) + self.log.append( + f"{label}: rejected {graph_rejects:,} candidates not in graph, {path_missing:,} candidates without a path, " + f"and {non_positive:,} candidates with non-positive net profit" + ) + self.log.append( + f"{label}: cached {len(leg_metrics):,} weighted travel legs from {graph_valid_summary['sources']:,} trade sources plus current location" + ) + self.log.append(f"{label}: best scored candidate {self.describe_trade(best_candidate)}") + + return profitable_matches, leg_metrics + + async def snipe_callback(self, updated_regions): + if not self.ship["location"]: + return if self.ship["location"] in self.graph.solar_systems: current_region = self.graph.solar_systems[self.ship["location"]]["region_id"] @@ -48,7 +440,7 @@ class Manager: current_region = self.graph.stations[self.ship["location"]]["region_id"] if current_region in updated_regions: - potential_snipe = snipe(self.market.orders, self.ship["cargo"], current_region) + potential_snipe = await snipe(self.market, self.ship["cargo"], current_region, self.ship["tax_rate"] / 100.0, self.ship["max_capital"], self.ship["snipe_profit_threshold"]) if potential_snipe: self.log.append("----------------") self.log.append(f"Snipe: {potential_snipe.item_name} orders for {potential_snipe.gross_profit} ISK") @@ -56,54 +448,87 @@ class Manager: print("snipe possible", potential_snipe.gross_profit, "ISK", potential_snipe.item_name) - open_market_window(potential_snipe.type_id) + await open_market_window(potential_snipe.type_id) - def update(self): + async def update(self): self.log = self.log[-100:] - location = get_location() - self.ship["location"] = location + location = await get_location() + if not self.ship["location"]: + self.ship["location"] = location + self.graph = NavigationGraph(self.ship) + else: + self.ship["location"] = location + + await self.reconcile_trade_confirmations() if len(self.route) == 1: # Only the last step is left, so we are at the destination self.log.append("Arrived at Destination") - self.route = [] + self.move_active_confirmations_to_pending() + self.clear_active_route() return for i, step in enumerate(self.route[:2]): if i != 0 and step["location_id"] == location: self.log.append("Arrived at " + step["location"]) - self.route = self.route[i:] + self.trim_active_route(i) + self.refresh_route_info() break if not self.updating_orders: self.updating_orders = True - def target_function(log): + async def target_function(): try: - updated_regions = self.market.update_orders() + updated_regions = await self.market.update_orders() if updated_regions: - log.append(f"Updated {len(updated_regions)} market regions") - self.snipe_callback(updated_regions) + self.log.append(f"Updated {len(updated_regions)} market regions") + await self.snipe_callback(updated_regions) finally: self.updating_orders = False - update_thread = threading.Thread(target=target_function, args=(self.log,)) - update_thread.start() + asyncio.create_task(target_function()) + + + async def create_route(self): + if self.ship["location"] is None: + self.ship["location"] = await get_location() + self.graph = NavigationGraph(self.ship) + if self.route or self.active_route_trades or self.active_trade_confirmations: + self.log.append("Discarded previous active route and its unconfirmed settlement tracking") + self.clear_active_route() - def create_route(self): - order_matches = arbitrage(self.market.orders, self.ship["cargo"]) - order_matches = pathfinder(order_matches, self.graph, self.ship) - self.route, self.route_info = route.route(order_matches, self.graph, self.ship) + route_start = perf_counter() + + order_matches, leg_metrics = await self.run_route_pass("Primary pass", PRIMARY_SHORTLIST) + if not order_matches: + self.log.append("Primary pass found no profitable candidates; retrying with a wider shortlist...") + order_matches, leg_metrics = await self.run_route_pass("Fallback pass", FALLBACK_SHORTLIST) + + self.log.append("Optimizing route...") + unique_triplets = route.count_unique_triplets(order_matches, self.ship["location"]) + self.log.append(f"Evaluating up to {min(unique_triplets, route.MAX_TRADES_TO_CONSIDER):,} unique triplets for route optimization") + stage_start = perf_counter() + ( + self.route, + self.route_info, + self.active_route_trades, + self.route_path, + self.route_path_offsets, + ) = route.route(order_matches, self.graph, self.ship, leg_metrics) + route_duration = perf_counter() - stage_start + total_duration = perf_counter() - route_start if self.route: - set_waypoints([r["location_id"] for r in self.route[1:]]) + self.active_trade_confirmations = self.build_trade_confirmations(self.active_route_trades) print(self.route) + self.log.append(f"Route complete in {total_duration:.2f}s ({route_duration:.2f}s optimizer)") else: - self.route = [] - self.route_info = {} + self.clear_active_route() print("No profitable route found yet.") + self.log.append(f"No profitable route found yet after {total_duration:.2f}s")
\ No newline at end of file |