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 WalletTransaction, get_location, get_wallet_transactions, open_market_window, set_waypoints as push_waypoints from market import Market 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": 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 "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 get_route(self): return self.route def get_route_info(self): return self.route_info def get_log(self): result = self.log self.log = [] return result 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"] else: current_region = self.graph.stations[self.ship["location"]]["region_id"] if current_region in updated_regions: 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") self.log.append("----------------") print("snipe possible", potential_snipe.gross_profit, "ISK", potential_snipe.item_name) await open_market_window(potential_snipe.type_id) async def update(self): self.log = self.log[-100:] 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.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.trim_active_route(i) self.refresh_route_info() break if not self.updating_orders: self.updating_orders = True async def target_function(): try: updated_regions = await self.market.update_orders() if updated_regions: self.log.append(f"Updated {len(updated_regions)} market regions") await self.snipe_callback(updated_regions) finally: self.updating_orders = False 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() 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: 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.clear_active_route() print("No profitable route found yet.") self.log.append(f"No profitable route found yet after {total_duration:.2f}s")