diff options
Diffstat (limited to 'backend/src/route.py')
| -rw-r--r-- | backend/src/route.py | 330 |
1 files changed, 270 insertions, 60 deletions
diff --git a/backend/src/route.py b/backend/src/route.py index 2bcae7e..8379e6d 100644 --- a/backend/src/route.py +++ b/backend/src/route.py @@ -1,10 +1,11 @@ +import heapq import logging -from dataclasses import dataclass, field +from collections import defaultdict from typing import Optional from tqdm import tqdm -from graph import NavigationGraph as Graph +from graph import NavigationGraph as Graph, LegMetrics from pathfinder import Ship from arbitrage import Trade from typing import Any @@ -16,8 +17,198 @@ MAX_TRADES_TO_CONSIDER = 20_000 MAX_CAPITAL = 100_000_000 Route = list[dict[str, Any]] +RoutePlan = tuple[Optional[Route], Optional[dict], list[Trade], list[int], list[int]] +LegMetricLookup = dict[tuple[int, int], Optional[LegMetrics]] -def route(trades: list[Trade], graph: Graph, ship: Ship) -> tuple[Optional[Route], Optional[dict]]: + +def outstanding_trades_for_locations(location_ids: list[int], trades: list[Trade]) -> list[Trade]: + remaining_locations = set(location_ids) + return [trade for trade in trades if trade.to_station in remaining_locations] + + +def calculate_peak_capital(location_ids: list[int], trades: list[Trade], ship_cost: float) -> float: + outstanding_trades = outstanding_trades_for_locations(location_ids, trades) + if not outstanding_trades: + return ship_cost + + remaining_locations = set(location_ids) + committed_capital = ship_cost + sum( + trade.from_price * trade.quantity + for trade in outstanding_trades + if trade.from_station not in remaining_locations + ) + peak_capital = committed_capital + + for location_id in location_ids: + committed_capital -= sum( + trade.from_price * trade.quantity + for trade in outstanding_trades + if trade.to_station == location_id + ) + committed_capital = max(ship_cost, committed_capital) + committed_capital += sum( + trade.from_price * trade.quantity + for trade in outstanding_trades + if trade.from_station == location_id + ) + peak_capital = max(peak_capital, committed_capital) + + return peak_capital + + +def summarize_graph_path(graph: Graph, path: list[int]) -> LegMetrics: + weight = 0.0 + transport_time = 0.0 + risk = 0.0 + + for source, target in zip(path, path[1:]): + edge = graph.graph[source][target] + weight += edge["weight"] + transport_time += edge["time"] + risk += edge["risk"] + + return LegMetrics(weight=weight, time=transport_time, risk=risk) + + +def summarize_location_sequence(graph: Graph, location_ids: list[int]) -> Optional[tuple[float, float]]: + if len(location_ids) < 2: + return 0.0, 0.0 + + transport_time = 0.0 + risk = 0.0 + + for source, target in zip(location_ids, location_ids[1:]): + if source == target: + continue + + if graph.graph.has_edge(source, target): + leg_metrics = LegMetrics( + weight=graph.graph[source][target]["weight"], + time=graph.graph[source][target]["time"], + risk=graph.graph[source][target]["risk"], + ) + else: + path = graph.shortest_path(source, target) + if path is None: + return None + leg_metrics = summarize_graph_path(graph, path) + + transport_time += leg_metrics.time + risk += leg_metrics.risk + + return transport_time, risk + + +def build_route_info( + location_ids: list[int], + trades: list[Trade], + ship: Ship, + transport_time: float, + risk: float, +) -> Optional[dict]: + outstanding_trades = outstanding_trades_for_locations(location_ids, trades) + if not outstanding_trades: + return None + + capital = calculate_peak_capital(location_ids, outstanding_trades, ship.get("ship_cost", 0.0)) + gross_profit = sum(trade.gross_profit for trade in outstanding_trades) + net_profit = gross_profit - risk * capital + profit_rate = net_profit / transport_time if transport_time > 0 else 0.0 + + return { + "profit_rate": profit_rate, + "risk": risk, + "capital": capital, + "transport_time": transport_time, + "gross_profit": gross_profit, + "net_profit": net_profit, + } + + +def extract_route_location_ids(route_steps: Route) -> list[int]: + return [ + step["location_id"] + for step in route_steps + if isinstance(step, dict) and isinstance(step.get("location_id"), int) + ] + + +def calculate_route_info( + route_steps: Route, + trades: list[Trade], + graph: Graph, + ship: Ship, + metric_path: Optional[list[int]] = None, +) -> Optional[dict]: + location_ids = [ + location_id + for location_id in extract_route_location_ids(route_steps) + ] + if not location_ids: + return None + + if metric_path is not None: + if len(metric_path) < 2: + transport_time = 0.0 + risk = 0.0 + else: + path_metrics = summarize_graph_path(graph, metric_path) + transport_time = path_metrics.time + risk = path_metrics.risk + else: + path_metrics = summarize_location_sequence(graph, location_ids) + if path_metrics is None: + return None + + transport_time, risk = path_metrics + + return build_route_info(location_ids, trades, ship, transport_time, risk) + + +def trade_capital_efficiency(trade: Trade) -> float: + invested_capital = trade.from_price * trade.quantity + return trade.gross_profit / invested_capital if invested_capital > 0 else 0.0 + + +def trade_profit_density(trade: Trade) -> float: + return trade.gross_profit / trade.cargo if trade.cargo > 0 else 0.0 + + +def count_unique_triplets(trades: list[Trade], ship_location: Optional[int]) -> int: + if ship_location is None: + return 0 + + return len({(ship_location, trade.from_station, trade.to_station) for trade in trades}) + + +def build_ranked_triplets(trades: list[Trade], ship_location: int) -> list[tuple[int, int, int]]: + ranked_triplets: list[tuple[int, int, int]] = [] + seen_triplets: set[tuple[int, int, int]] = set() + + for trade in sorted(trades, key=trade_capital_efficiency, reverse=True): + triplet = (ship_location, trade.from_station, trade.to_station) + if triplet in seen_triplets: + continue + + seen_triplets.add(triplet) + ranked_triplets.append(triplet) + + return ranked_triplets + + +def build_station_pair_index(trades: list[Trade]) -> dict[tuple[int, int], list[Trade]]: + trades_by_station_pair: dict[tuple[int, int], list[Trade]] = defaultdict(list) + + for trade in trades: + trades_by_station_pair[(trade.from_station, trade.to_station)].append(trade) + + for station_pair_trades in trades_by_station_pair.values(): + station_pair_trades.sort(key=trade_profit_density, reverse=True) + + return dict(trades_by_station_pair) + + +def route(trades: list[Trade], graph: Graph, ship: Ship, leg_metrics: Optional[LegMetricLookup] = None) -> RoutePlan: """ Given a list of trades, find the best route starting from the ship's location. @@ -29,92 +220,109 @@ def route(trades: list[Trade], graph: Graph, ship: Ship) -> tuple[Optional[Route Returns: A tuple containing the best route and route information. """ - trades.sort( - key=lambda trade: trade.gross_profit / (trade.from_price * trade.quantity) if trade.from_price * trade.quantity > 0 else 0, - reverse=True - ) + if not trades: + return None, None, [], [], [] + + ranked_triplets = build_ranked_triplets(trades, ship["location"]) + trades_by_station_pair = build_station_pair_index(trades) + + if leg_metrics is None: + leg_metrics = graph.build_trade_leg_metrics(ship["location"], trades) best_profit_rate = 0.0 best_route: Optional[Route] = None best_route_info: Optional[dict] = None + best_route_stations: Optional[tuple[int, int, int]] = None + best_optimized_trades: Optional[list[Trade]] = None - seen_routes = set() - - for trade in tqdm(trades[:MAX_TRADES_TO_CONSIDER], desc="Evaluating trades"): - route_stations = (ship["location"], trade.from_station, trade.to_station) - - if route_stations in seen_routes: + for route_stations in tqdm(ranked_triplets[:MAX_TRADES_TO_CONSIDER], desc="Evaluating triplets"): + optimized_trades = select_trades(route_stations, trades_by_station_pair, ship) + if not optimized_trades: continue - seen_routes.add(route_stations) - - optimized_trades = select_trades(list(route_stations), trades, ship) - total_profit = sum(t.gross_profit for t in optimized_trades) - if not optimized_trades or total_profit <= 0: - continue + source_leg = leg_metrics.get((route_stations[0], route_stations[1])) + destination_leg = leg_metrics.get((route_stations[1], route_stations[2])) - try: - path = graph.shortest_path(ship["location"], trade.from_station)[:-1] - path += graph.shortest_path(trade.from_station, trade.to_station) - except Exception as e: - logger.error(f"Error finding path: {e}") + if source_leg is None or destination_leg is None: continue - risk = sum(graph.graph[u][v]['risk'] for u, v in zip(path, path[1:])) - transport_time = sum(graph.graph[u][v]['time'] for u, v in zip(path, path[1:])) - - capital = ship.get('ship_cost', 0.0) + sum(t.from_price * t.quantity for t in optimized_trades) + risk = source_leg.risk + destination_leg.risk + transport_time = source_leg.time + destination_leg.time - if transport_time == 0: - logger.warning("Transport time is zero, skipping this route to avoid division by zero.") + candidate_route_info = build_route_info(list(route_stations), optimized_trades, ship, transport_time, risk) + if candidate_route_info is None: continue - net_profit = total_profit - risk * capital - profit_rate = net_profit / transport_time + profit_rate = candidate_route_info["profit_rate"] if profit_rate > best_profit_rate: logger.info( f"New best profit rate: {profit_rate * 3600:.2f} per hour, " - f"Risk: {risk:.4f}, Capital: {capital:.2f}, " - f"Transport time: {transport_time:.2f} sec, Gross profit: {total_profit:.2f}" + f"Risk: {risk:.4f}, Capital: {candidate_route_info['capital']:.2f}, " + f"Transport time: {transport_time:.2f} sec, Gross profit: {candidate_route_info['gross_profit']:.2f}" ) - formatted_route = graph.formatted_route(path) - final_route = set_actions(formatted_route, optimized_trades) - best_route = final_route - best_route_info = { - "profit_rate": profit_rate, - "risk": risk, - "capital": capital, - "transport_time": transport_time, - "gross_profit": total_profit, - "net_profit": net_profit, - } + best_route_info = candidate_route_info best_profit_rate = profit_rate + best_route_stations = route_stations + best_optimized_trades = optimized_trades + + if best_route_stations is None or best_optimized_trades is None: + return None, None, [], [], [] + + path = reconstruct_route_path(graph, best_route_stations) + if path is None: + logger.error("Unable to reconstruct best route path for stations %s", best_route_stations) + return None, None, [], [], [] + + formatted_route, raw_offsets = graph.formatted_route_with_offsets(path) + best_route = set_actions(formatted_route, best_optimized_trades) + best_route_info = calculate_route_info(best_route, best_optimized_trades, graph, ship, metric_path=path) or best_route_info - return best_route, best_route_info + return best_route, best_route_info, best_optimized_trades, path, raw_offsets -def select_trades(stations: list[int], trades: list[Trade], ship: Ship) -> list[Trade]: +def reconstruct_route_path(graph: Graph, route_stations: tuple[int, int, int]) -> Optional[list[int]]: + source_path = graph.shortest_path(route_stations[0], route_stations[1]) + if source_path is None: + return None + + destination_path = graph.shortest_path(route_stations[1], route_stations[2]) + if destination_path is None: + return None + + path = source_path[:-1] + path.extend(destination_path) + return path + + +def select_trades(route_stations: tuple[int, int, int], trades_by_station_pair: dict[tuple[int, int], list[Trade]], ship: Ship) -> list[Trade]: """ - Given a list of stations, find the optimal trades to add. + Given a triplet of stations, find the optimal trades to add. Args: - stations: List of station IDs. - trades: List of Trade instances. + route_stations: Ordered station triplet for the route. + trades_by_station_pair: Trade instances grouped and pre-sorted by ordered station pair. ship: Ship parameters. Returns: A list of optimized Trade instances. """ - filtered_trades = [ - trade for trade in trades - if trade.from_station in stations and trade.to_station in stations and - stations.index(trade.from_station) < stations.index(trade.to_station) + candidate_pairs: list[tuple[int, int]] = [] + for index, from_station in enumerate(route_stations): + for to_station in route_stations[index + 1:]: + station_pair = (from_station, to_station) + if station_pair not in candidate_pairs: + candidate_pairs.append(station_pair) + + pair_groups = [ + trades_by_station_pair[station_pair] + for station_pair in candidate_pairs + if station_pair in trades_by_station_pair ] - filtered_trades.sort( - key=lambda trade: trade.gross_profit / trade.cargo if trade.cargo > 0 else 0, - reverse=True - ) + if not pair_groups: + return [] + + filtered_trades = heapq.merge(*pair_groups, key=trade_profit_density, reverse=True) used_orders = {} optimized_trades = [] @@ -154,7 +362,7 @@ def select_trades(stations: list[int], trades: list[Trade], ship: Ship) -> list[ to_price=trade.to_price, quantity=max_quantity, cargo=trade.item_cargo_volume * max_quantity, - gross_profit=(trade.to_price - trade.from_price) * max_quantity, + gross_profit=trade.gross_profit * (max_quantity / trade.quantity) if trade.quantity > 0 else 0.0, net_profit=None # Will be calculated later if needed ) @@ -198,6 +406,8 @@ def set_actions(route: Route, trades: list[Trade]) -> Route: }) for step in route: - step["actions"].sort(key=lambda action: (action["action_type"], action["item"], action["price"])) + step["actions"].sort( + key=lambda action: ({"sell": 0, "buy": 1}[action["action_type"]], action["item"], action["price"]) + ) return route |