1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
|
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
]
|
