Merge pull request #104 from JinBoatus1/main

post-trade analysis

Author: Allan-Feng

dashboard/backend/api/routers/backtests.py

@@ -197,13 +197,15 @@ def run_backtest_background(
     strategy_prompt: Optional[str] = None,
     model: Optional[str] = None,
     pipeline: Optional[List[Dict[str, Any]]] = None,
+    agent_id: Optional[str] = None,
 ):
     """Run backtest in background thread."""
     global backtest_status, backtest_session_id
 
     strategy_prompt_path = None
     pipeline_path = None
     progress_file = None
+    live_run_id = None
     try:
         import subprocess
         import sys
@@ -304,33 +306,63 @@ def run_backtest_background(
             print(f"✅ Backtest completed. Found {len(runs)} runs in database.", flush=True)
             if len(runs) > 0:
                 print(f"   Latest run IDs: {[r['run_id'] for r in runs[:3]]}", flush=True)
+            _maybe_writeback_adapted_pipeline(agent_id, live_run_id)
     except Exception as e:
         backtest_status["error"] = str(e)
         print(f"❌ Backtest exception: {e}", flush=True)
     finally:
+        import os
+
         backtest_status["running"] = False
         backtest_status["started_at"] = None
         backtest_status["live_run_id"] = None
+        backtest_status["progress_file"] = None
         if progress_file:
             try:
-                import os
-                os.remove(progress_file)
+                Path(progress_file).unlink(missing_ok=True)
             except OSError:
                 pass
-        backtest_status["progress_file"] = None
         if strategy_prompt_path:
             try:
-                import os
                 os.remove(strategy_prompt_path)
             except OSError:
                 pass
         if pipeline_path:
             try:
-                import os
                 os.remove(pipeline_path)
             except OSError:
                 pass
-        print(f"✋ Backtest background thread finished", flush=True)
+        print("✋ Backtest background thread finished", flush=True)
+
+
+def _maybe_writeback_adapted_pipeline(agent_id: Optional[str], run_id: Optional[str]) -> None:
+    """Persist post-trade adapted pipeline back onto the agent row."""
+    if not agent_id or not run_id:
+        return
+    run = db.get_run(run_id)
+    if not run:
+        return
+    metadata = run.get("metadata")
+    if isinstance(metadata, str):
+        try:
+            metadata = json.loads(metadata)
+        except json.JSONDecodeError:
+            metadata = None
+    if not isinstance(metadata, dict):
+        return
+    adaptations = metadata.get("prompt_adaptations")
+    final_pipeline = metadata.get("final_pipeline")
+    if not adaptations or not isinstance(final_pipeline, list) or not final_pipeline:
+        return
+    try:
+        agent_service.update_agent(agent_id, pipeline=final_pipeline)
+        print(
+            f"✅ Wrote adapted pipeline back to agent {agent_id} "
+            f"({len(adaptations)} adaptation day(s))",
+            flush=True,
+        )
+    except Exception as exc:
+        print(f"⚠️  Could not write adapted pipeline to agent {agent_id}: {exc}", flush=True)
 
 class BacktestRunRequest(BaseModel):
     """Optional JSON body for POST /backtest/run.
@@ -535,7 +567,7 @@ async def run_backtest_endpoint(
     print(f"🧵 Starting background thread for backtest", flush=True)
     thread = threading.Thread(
         target=run_backtest_background,
-        args=(start_date, end_date, session_id, strategy_prompt, model, pipeline),
+        args=(start_date, end_date, session_id, strategy_prompt, model, pipeline, agent_id),
         daemon=True
     )
     thread.start()

dashboard/backend/app.py

@@ -349,5 +349,7 @@ async def serve_image(file_name: str):
     # Canonical startup is ``uvicorn dashboard.backend.app:app``. This direct
     # invocation is a deprecated compatibility path; reference the app by its
     # canonical import string so the reloader resolves the same module identity.
+    import os
     import uvicorn
-    uvicorn.run("dashboard.backend.app:app", host="0.0.0.0", port=8000, reload=True)
+    port = int(os.environ.get("PORT", "8000"))
+    uvicorn.run("dashboard.backend.app:app", host="0.0.0.0", port=port, reload=True)

dashboard/backend/domain/backtesting/engine.py

@@ -40,6 +40,13 @@
     default_model_name,
     make_llm_client,
 )
+from dashboard.backend.infrastructure.llm.pipeline_runner import (
+    is_last_bar_of_trading_day,
+    recombine_pipeline,
+    run_post_trade_analysis,
+    split_pipeline,
+    trading_day_key,
+)
 
 
 class HourlyBacktester:
@@ -66,6 +73,10 @@ def __init__(self, start_date: str, end_date: str, session_id: str = "legacy-dem
         self.strategy_prompt = (strategy_prompt or "").strip() or None
         # Optional sub-agent pipeline (when set, overrides strategy_prompt).
         self.pipeline = pipeline if pipeline else None
+        self.initial_pipeline = (
+            json.loads(json.dumps(self.pipeline)) if self.pipeline else None
+        )
+        self.prompt_adaptations: List[Dict] = []
         # Model id; defaults to the gateway-appropriate slug (CommonStack vs native).
         self.model = model or default_model_name()
         self.live_run_id = (live_run_id or "").strip() or None
@@ -178,10 +189,66 @@ def _llm_run_metadata(self) -> Optional[Dict]:
         LLM_MAX_OUTPUT_TOKENS is an env knob that changes a run's spend and
         response truncation; recording the EFFECTIVE value (post defensive
         parse) makes runs auditable after the env changes. Rule-based runs
-        record nothing."""
-        if not self.use_llm:
-            return None
-        return {"llm_max_output_tokens": llm_harness.DEFAULT_MAX_OUTPUT_TOKENS}
+        record nothing unless post-trade adaptations were produced."""
+        meta: Dict = {}
+        if self.use_llm:
+            meta["llm_max_output_tokens"] = llm_harness.DEFAULT_MAX_OUTPUT_TOKENS
+        if self.prompt_adaptations:
+            meta["prompt_adaptations"] = self.prompt_adaptations
+        if self.initial_pipeline is not None:
+            meta["initial_pipeline"] = self.initial_pipeline
+        if self.pipeline is not None:
+            meta["final_pipeline"] = self.pipeline
+        return meta or None
+
+    def _current_equity(self, manager: PortfolioManager) -> float:
+        if manager.equity_history:
+            return float(manager.equity_history[-1].get("equity") or manager.cash)
+        return float(manager.cash)
+
+    def _run_daily_post_trade(
+        self,
+        *,
+        manager: PortfolioManager,
+        day_episode: Dict,
+        post_trade_steps: List[Dict],
+    ) -> None:
+        """Run once-per-day post-trade analysis and mutate ``self.pipeline``."""
+        if not post_trade_steps or not self.use_llm or not self.llm_client:
+            return
+        if not day_episode.get("trading_day"):
+            return
+
+        decision_steps, _ = split_pipeline(self.pipeline)
+        start_eq = float(day_episode.get("day_start_equity") or 0)
+        end_eq = self._current_equity(manager)
+        day_return = ((end_eq - start_eq) / start_eq) if start_eq else 0.0
+        trade_start = int(day_episode.get("trade_start_index") or 0)
+        day_trades = manager.trades[trade_start:]
+
+        episode_context = {
+            "trading_day": day_episode.get("trading_day"),
+            "day_start_equity": start_eq,
+            "day_end_equity": end_eq,
+            "day_return": day_return,
+            "trade_count": len(day_trades),
+            "trades": day_trades,
+            "latest_step_outputs": day_episode.get("latest_step_outputs") or [],
+        }
+
+        patched, record, (in_tok, out_tok), calls = run_post_trade_analysis(
+            self.llm_client,
+            post_trade_steps=post_trade_steps,
+            episode_context=episode_context,
+            decision_pipeline=decision_steps,
+            model=self.model,
+        )
+        manager.input_tokens += in_tok
+        manager.output_tokens += out_tok
+        manager.llm_calls += calls
+        if record:
+            self.prompt_adaptations.append(record)
+        self.pipeline = recombine_pipeline(patched, post_trade_steps)
 
     def run_agent_backtest(self) -> Tuple[str, List[Dict]]:
         """Run backtest with agent making hourly decisions."""
@@ -192,6 +259,12 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]:
         llm_model = "rule-based"  # Default
 
         manager = PortfolioManager(initial_capital=INITIAL_CAPITAL)
+        _decision_steps, post_trade_steps = split_pipeline(self.pipeline)
+        if post_trade_steps:
+            print(
+                f"   Post-trade analysis: {len(post_trade_steps)} step(s), "
+                "once per trading day\n"
+            )
 
         # Get all timestamps
         all_timestamps = set()
@@ -251,9 +324,25 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]:
                         price_cache[symbol][timestamp] = last_price
 
         print("   ✅ Cache ready\n")
+
+        day_episode: Dict = {
+            "trading_day": None,
+            "day_start_equity": None,
+            "trade_start_index": 0,
+            "latest_step_outputs": [],
+        }
 
         # Hourly loop
         for i, timestamp in enumerate(all_timestamps):
+            day_key = trading_day_key(timestamp)
+            if day_episode["trading_day"] != day_key:
+                day_episode = {
+                    "trading_day": day_key,
+                    "day_start_equity": self._current_equity(manager),
+                    "trade_start_index": len(manager.trades),
+                    "latest_step_outputs": [],
+                }
+
             # Get market data for this hour (real data when available)
             market_data = {}
             for symbol in DJIA_30:
@@ -281,6 +370,8 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]:
                 llm_calls_count += 1  # Track that LLM was used
                 if llm_calls_count == 1:  # Set on first call
                     llm_model = self.model
+                if manager.last_pipeline_step_outputs:
+                    day_episode["latest_step_outputs"] = manager.last_pipeline_step_outputs
             else:
                 decision = manager.make_trading_decision(state)
 
@@ -290,6 +381,13 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]:
             # Update equity (uses forward-filled prices for smooth valuation)
             manager.update_equity(market_data, price_cache, timestamp)
             self._publish_live_progress(i + 1, total_steps, manager)
+
+            if post_trade_steps and is_last_bar_of_trading_day(all_timestamps, i):
+                self._run_daily_post_trade(
+                    manager=manager,
+                    day_episode=day_episode,
+                    post_trade_steps=post_trade_steps,
+                )
 
             # Progress
             if (i + 1) % 100 == 0:
@@ -345,6 +443,8 @@ def run_agent_backtest(self) -> Tuple[str, List[Dict]]:
         print(f"     • LLM Calls: {llm_calls_count}")
         print(f"     • Tokens: {manager.input_tokens:,} in / {manager.output_tokens:,} out (est. cost ${est_cost:.4f})")
         print(f"     • Trades: {len(manager.trades)}")
+        if self.prompt_adaptations:
+            print(f"     • Post-trade adaptations: {len(self.prompt_adaptations)} day(s)")
         print(f"     • Final: ${final_eq:,.0f}")
         print(f"     • Return: {total_return*100:+.2f}%\n")
 

dashboard/backend/domain/backtesting/portfolio_manager.py

@@ -70,7 +70,8 @@ def __init__(self, initial_capital: float = 100000):
         self.llm_decisions = 0    # steps the model actually drove (H6 coverage)
         self.input_tokens = 0
         self.output_tokens = 0
-    
+        # Latest decision-pipeline step outputs (for daily post-trade analysis).
+        self.last_pipeline_step_outputs = []    
     def get_portfolio_state(self, market_data: Dict[str, pd.Series], price_cache: Dict = None, timestamp = None) -> Dict:
         """Get current portfolio state with market indicators.
 
@@ -287,15 +288,18 @@ def _trend_score(sig: Dict) -> float:
 
             if pipeline:
                 print(f"   Sub-agent pipeline: {len(pipeline)} step(s)")
-                decision, (input_delta, output_delta), pipeline_calls = run_pipeline_decision(
-                    llm_client,
-                    pipeline=pipeline,
-                    market_snapshot=market_snapshot,
-                    model=model,
+                decision, (input_delta, output_delta), pipeline_calls, step_outputs = (
+                    run_pipeline_decision(
+                        llm_client,
+                        pipeline=pipeline,
+                        market_snapshot=market_snapshot,
+                        model=model,
+                    )
                 )
                 self.input_tokens += input_delta
                 self.output_tokens += output_delta
                 self.llm_calls += pipeline_calls
+                self.last_pipeline_step_outputs = step_outputs or []
                 if decision is None:
                     print("   Falling back to rule-based logic")
                     return self.make_trading_decision(portfolio_state)