feat(core): implement wallet trade tracking and sqlite market caching

- Transitioned primary ESI networking and dataset initialization to `asyncio` and `aiohttp` for non-blocking I/O.
- Introduced `aiosqlite` utilizing an in-memory database to significantly speed up arbitrage querying.
- Implemented character wallet tracking (added `esi-wallet.read_character_wallet.v1` scope) to automatically deduct filled market volumes internally instead of waiting for full region polling.
- Exposed `/settings` endpoint allowing dynamic tweaking of threshold variables (cargo size, limits, tax rate, ship speeds) directly from the React UI.
- Decoupled auto-waypoint submission during route planning; users now push routes selectively via a new "Set in Game" frontend action and `/set_waypoints` backend logic.
- Optimized order evaluations and triplet generation limits through `LegMetrics` to lower CPU footprint drastically on large dataset passes.
- Migrated React frontend completely to Server-Sent Events (SSE), cleanly ditching HTTP polling.

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