import logging from dataclasses import dataclass from typing import Optional from tqdm import tqdm from arbitrage import Trade from graph import NavigationGraph as Graph, LegMetrics logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) LegMetricLookup = dict[tuple[int, int], Optional[LegMetrics]] @dataclass class Ship: location: int cargo: int max_warp_speed: float max_subwarp_speed: float gankers_areas: list[str] player_cost: float risk_cost: float def pathfinder(order_matches: list[Trade], graph: Graph, ship: Ship, leg_metrics: Optional[LegMetricLookup] = None) -> list[Trade]: """ Processes order matches by filtering and calculating net profit. Args: order_matches: List of trade opportunities. graph: Navigation graph. ship: Ship parameters. Returns: A list of filtered and annotated trade opportunities. """ order_matches = filter_order_matches_not_in_graph(order_matches, graph) logger.info("After filtering order matches not in graph: %d", len(order_matches)) if leg_metrics is None: leg_metrics = graph.build_trade_leg_metrics(ship["location"], order_matches) order_matches = calculate_net_profit(order_matches, ship["location"], leg_metrics) order_matches = filter_order_matches_low_profit(order_matches) logger.info("After filtering low profit order matches: %d", len(order_matches)) return order_matches def filter_order_matches_not_in_graph(order_matches: list[Trade], graph: Graph) -> list[Trade]: """ Filters out order matches where either the from_station or to_station is not in the graph. Args: order_matches: List of trade opportunities. graph: Navigation graph. Returns: Filtered list of trade opportunities. """ return [ order_match for order_match in order_matches if order_match.from_station in graph.graph.nodes and order_match.to_station in graph.graph.nodes ] def calculate_net_profit(order_matches: list[Trade], ship_location: int, leg_metrics: LegMetricLookup) -> list[Trade]: """ Calculates the net profit for each order match by subtracting transport cost from gross profit. Args: order_matches: List of trade opportunities. graph: Navigation graph. ship: Ship parameters. Returns: List of trade opportunities with net_profit calculated. """ logger.info("Calculating estimated profits") for order_match in tqdm(order_matches, desc="Scoring trades"): try: # Calculate transport cost from ship location to from_station cost_to_source = leg_metrics.get((ship_location, order_match.from_station)) # Calculate transport cost from from_station to to_station cost_to_destination = leg_metrics.get((order_match.from_station, order_match.to_station)) # Check if paths exist if cost_to_source is None or cost_to_destination is None: logger.warning(f"No path found for order match: {order_match}") order_match.net_profit = None continue total_transport_cost = cost_to_source.weight + cost_to_destination.weight order_match.net_profit = order_match.gross_profit - total_transport_cost except Exception as e: logger.error(f"Error calculating net profit for order match {order_match}: {e}") order_match.net_profit = None return order_matches def filter_order_matches_low_profit(order_matches: list[Trade]) -> list[Trade]: """ Filters out order matches with non-positive net profit. Args: order_matches: List of trade opportunities with net_profit calculated. Returns: Filtered list of profitable trade opportunities. """ return [ order_match for order_match in order_matches if order_match.net_profit and order_match.net_profit > 0 ]