feat: LangGraph best-practices refactor (Sprint 8 Epics 2-3-4)

- Fix state mutation anti-pattern: all 4 workflow agent nodes + debug
wrappers now return partial dicts instead of mutating state
- Add Annotated reducers (operator.add) to candidates/candidate_scores
list fields in AgentState
- Replace deprecated set_entry_point / set_conditional_entry_point with
START edges across agent.py, whale_hunter.py, workflow.py
- Refactor gatekeeper routing to Command pattern, removing separate
check_status functions in agent.py and whale_hunter.py
- Add InMemorySaver checkpointer to all 3 graph compile() calls with
thread_id config on every invoke/ainvoke
- Add RetryPolicy(max_attempts=3) to all graph nodes and
recursion_limit=30 to all invoke calls
- Add InvestmentVerdict Pydantic model (src/models/verdict.py) with
structured output via with_structured_output, graceful fallback to
plain LLM
- Simplify portfolio_tracker verdict parsing when structured_verdict is
provided
- Clean up requirements.txt: LangChain 1.0 LTS range pinning, remove
duplicates and deprecated langchain-classic
- Refactor whale_hunter to parallel region fan-out via Send API with
GlobalHunterState orchestrator graph
- Include VPS API layer (vps/) and memory/portfolio VPS integration

Made-with: Cursor

.github/workflows/hunter.yml

@@ -45,6 +45,8 @@ jobs:
           LANGCHAIN_API_KEY: ${{ secrets.LANGCHAIN_API_KEY }}
           LANGCHAIN_PROJECT: primogreedy
           LANGSMITH_WORKSPACE_ID: ${{ secrets.LANGSMITH_WORKSPACE_ID }}
+          VPS_API_URL: ${{ secrets.VPS_API_URL }}
+          VPS_API_KEY: ${{ secrets.VPS_API_KEY }}
         run: PYTHONPATH=. python src/whale_hunter.py
 
       # 🚨 CRITICAL NEW STEP: Save the memory file safely without crashing

app.py

@@ -90,7 +90,7 @@ async def main(message: cl.Message):
     elif " " in user_input:
         await cl.Message(content="Consulting Senior Broker...").send()
         try:
-            config = {"configurable": {"thread_id": "ui_session"}}
+            config = {"configurable": {"thread_id": "ui_session"}, "recursion_limit": 30}
             result = await app.ainvoke(
                 {"ticker": user_input, "retry_count": 0, "manual_search": False},
                 config=config,
@@ -112,7 +112,7 @@ async def main(message: cl.Message):
     for ticker in tickers:
         await cl.Message(content=f"--- **Processing:** {ticker} ---").send()
         try:
-            config = {"configurable": {"thread_id": "ui_session"}}
+            config = {"configurable": {"thread_id": "ui_session"}, "recursion_limit": 30}
             result = await app.ainvoke(
                 {"ticker": ticker, "retry_count": 0, "manual_search": True},
                 config=config,

requirements.txt

@@ -17,10 +17,10 @@ httpx-sse==0.4.3
 idna==3.11
 jsonpatch==1.33
 jsonpointer==3.0.0
-langchain-classic==1.0.1
-langchain-community==0.4.1
-langchain-core==1.2.16
-langchain-text-splitters==1.1.1
+langchain>=1.0,<2.0
+langchain-community>=0.4.0,<0.5.0
+langchain-core>=1.0,<2.0
+langchain-text-splitters>=1.0,<2.0
 langsmith==0.7.9
 marshmallow==3.26.2
 multidict==6.7.1
@@ -46,11 +46,9 @@ uuid_utils==0.14.1
 xxhash==3.6.0
 yarl==1.22.0
 zstandard==0.25.0
-finnhub-python
-langchain-community
-langchain-openai
+langchain-openai>=0.3
+langgraph>=1.0,<2.0
 chainlit
-langgraph
 resend
 yfinance
 matplotlib

src/agent.py

@@ -10,7 +10,10 @@ import random
 import time
 import yfinance as yf
 import matplotlib.pyplot as plt
-from langgraph.graph import StateGraph, END
+from typing import Literal
+from langgraph.graph import StateGraph, START, END
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.types import Command, RetryPolicy
 
 from src.llm import get_llm, invoke_with_fallback
 from src.finance_tools import (
@@ -128,18 +131,31 @@ def scout_node(state):
     return {"ticker": ticker, "manual_search": False}
 
 
-def gatekeeper_node(state):
-    """Validate candidate with financial health checks."""
+def _gatekeeper_route(state, update) -> str:
+    """Decide where gatekeeper should route based on state + update."""
+    if state.get("manual_search"):
+        return "analyst"
+    if update.get("status") == "PASS":
+        return "analyst"
+    new_retries = update.get("retry_count", state.get("retry_count", 0))
+    if new_retries >= MAX_RETRIES:
+        return "analyst"
+    return "scout"
+
+
+def gatekeeper_node(state) -> Command[Literal["analyst", "scout"]]:
+    """Validate candidate with financial health checks. Routes via Command."""
     ticker = state.get("ticker", "NONE")
     retries = state.get("retry_count", 0)
 
     if ticker == "NONE":
-        return {
+        update = {
             "is_small_cap": False,
             "status": "FAIL",
             "retry_count": retries + 1,
             "financial_data": {"reason": "Scout found no readable ticker."},
         }
+        return Command(update=update, goto=_gatekeeper_route(state, update))
 
     mark_ticker_seen(ticker)
 
@@ -168,7 +184,7 @@ def gatekeeper_node(state):
         chart_bytes = generate_chart(ticker)
 
         if price > MAX_PRICE_PER_SHARE:
-            return {
+            update = {
                 "market_cap": mkt_cap,
                 "is_small_cap": False,
                 "status": "FAIL",
@@ -178,9 +194,10 @@ def gatekeeper_node(state):
                 "final_report": f"Price ${price:.2f} exceeds ${MAX_PRICE_PER_SHARE} limit.",
                 "chart_data": chart_bytes,
             }
+            return Command(update=update, goto=_gatekeeper_route(state, update))
 
         if not (MIN_MARKET_CAP < mkt_cap < MAX_MARKET_CAP):
-            return {
+            update = {
                 "market_cap": mkt_cap,
                 "is_small_cap": False,
                 "status": "FAIL",
@@ -190,10 +207,11 @@ def gatekeeper_node(state):
                 "final_report": f"Market Cap ${mkt_cap:,.0f} is outside the $10M-$300M range.",
                 "chart_data": chart_bytes,
             }
+            return Command(update=update, goto=_gatekeeper_route(state, update))
 
         health = check_financial_health(ticker, lean_info)
         if health["status"] == "FAIL":
-            return {
+            update = {
                 "market_cap": mkt_cap,
                 "is_small_cap": False,
                 "status": "FAIL",
@@ -203,8 +221,9 @@ def gatekeeper_node(state):
                 "final_report": f"**GATEKEEPER REJECT:** {health['reason']}",
                 "chart_data": chart_bytes,
             }
+            return Command(update=update, goto=_gatekeeper_route(state, update))
 
-        return {
+        update = {
             "market_cap": mkt_cap,
             "is_small_cap": True,
             "status": "PASS",
@@ -212,15 +231,17 @@ def gatekeeper_node(state):
             "financial_data": lean_info,
             "chart_data": chart_bytes,
         }
+        return Command(update=update, goto="analyst")
 
     except Exception as exc:
         logger.error("Gatekeeper error for %s: %s", ticker, exc)
-        return {
+        update = {
             "is_small_cap": False,
             "status": "FAIL",
             "retry_count": retries + 1,
             "financial_data": {"reason": f"API Error: {exc}"},
         }
+        return Command(update=update, goto=_gatekeeper_route(state, update))
 
 
 def analyst_node(state):
@@ -299,22 +320,33 @@ def analyst_node(state):
     )
 
     try:
-        verdict = invoke_with_fallback(prompt, run_name="analyst_node")
-        record_paper_trade(ticker, price, verdict, source="Chainlit UI")
+        from src.models.verdict import InvestmentVerdict
+        structured_llm = get_llm().with_structured_output(InvestmentVerdict)
+        result = structured_llm.invoke(prompt)
+        verdict = result.to_report()
+        record_paper_trade(ticker, price, verdict, source="Chainlit UI",
+                           structured_verdict=result.verdict)
     except Exception as exc:
-        logger.error("LLM analysis failed for %s: %s", ticker, exc)
-        verdict = f"Strategy: {strategy}\nLLM analysis unavailable: {exc}"
+        logger.warning("Structured output failed for %s, falling back to plain LLM: %s", ticker, exc)
+        try:
+            verdict = invoke_with_fallback(prompt, run_name="analyst_node")
+            record_paper_trade(ticker, price, verdict, source="Chainlit UI")
+        except Exception as exc2:
+            logger.error("LLM analysis failed for %s: %s", ticker, exc2)
+            verdict = f"Strategy: {strategy}\nLLM analysis unavailable: {exc2}"
 
     return {"final_verdict": verdict, "final_report": verdict, "chart_data": chart_bytes}
 
 
 # --- GRAPH ---
 
+_api_retry = RetryPolicy(max_attempts=3, initial_interval=2.0)
+
 workflow = StateGraph(AgentState)
-workflow.add_node("chat", chat_node)
-workflow.add_node("scout", scout_node)
-workflow.add_node("gatekeeper", gatekeeper_node)
-workflow.add_node("analyst", analyst_node)
+workflow.add_node("chat", chat_node, retry=_api_retry)
+workflow.add_node("scout", scout_node, retry=_api_retry)
+workflow.add_node("gatekeeper", gatekeeper_node, retry=_api_retry)
+workflow.add_node("analyst", analyst_node, retry=_api_retry)
 
 
 def initial_routing(state):
@@ -325,29 +357,9 @@ def initial_routing(state):
     return "scout"
 
 
-workflow.set_conditional_entry_point(
-    initial_routing,
-    {"chat": "chat", "scout": "scout"},
-)
-
-
-def check_status(state):
-    if state.get("manual_search"):
-        return "analyst"
-    if state.get("status") == "PASS":
-        return "analyst"
-    if state.get("retry_count", 0) >= MAX_RETRIES:
-        return "analyst"
-    return "scout"
-
-
+workflow.add_conditional_edges(START, initial_routing, ["chat", "scout"])
 workflow.add_edge("chat", END)
 workflow.add_edge("scout", "gatekeeper")
-workflow.add_conditional_edges(
-    "gatekeeper",
-    check_status,
-    {"analyst": "analyst", "scout": "scout"},
-)
 workflow.add_edge("analyst", END)
 
-app = workflow.compile()
+app = workflow.compile(checkpointer=InMemorySaver())

src/agents/data_collection_agent.py

@@ -46,35 +46,23 @@ async def collect_data(symbol: str, analysis_date: Optional[str] = None) -> Dict
         }
 
 
-async def data_collection_agent_node(state: AgentState) -> AgentState:
-    """
-    LangGraph node for data collection.
-    
-    Args:
-        state: Current workflow state
-        
-    Returns:
-        Updated state with data collection results
-    """
+async def data_collection_agent_node(state: AgentState) -> dict:
+    """LangGraph node for data collection. Returns partial state updates."""
     try:
-        # Get first symbol from state
         symbol = state['symbols'][0] if state['symbols'] else 'AAPL'
         analysis_date = state['analysis_date']
         
-        # Collect data with analysis date
         result = await collect_data(symbol, analysis_date)
         
-        # Update state
-        state['data_collection_results'] = result
-        state['current_step'] = 'data_collection_complete'
-        
+        updates: dict = {
+            "data_collection_results": result,
+            "current_step": "data_collection_complete",
+        }
         if not result['success']:
-            state['error'] = result.get('error', 'Data collection failed')
-            
-        return state
+            updates["error"] = result.get('error', 'Data collection failed')
+
+        return updates
         
     except Exception as e:
         print(f"Data collection node error: {e}")
-        state['error'] = str(e)
-        state['current_step'] = 'error'
-        return state 
+        return {"error": str(e), "current_step": "error"}

src/agents/news_intelligence_agent.py

@@ -331,33 +331,25 @@ async def extract_nlp_features(
         return None
 
 
-async def news_intelligence_agent_node(state: AgentState) -> AgentState:
-    """
-    LangGraph node for news intelligence with complete PrimoGPT workflow.
-    """
+async def news_intelligence_agent_node(state: AgentState) -> dict:
+    """LangGraph node for news intelligence. Returns partial state updates."""
     try:
-        # Get symbol, analysis_date and technical data from previous agents
         symbol = state['symbols'][0] if state['symbols'] else 'AAPL'
         analysis_date = state['analysis_date']
         technical_data = state.get('technical_analysis_results')
-        
-        # Get company data from data collection results
         company_data = state.get('data_collection_results')
         
-        # Perform complete news analysis with company context
         result = await analyze_news(symbol, analysis_date, technical_data, company_data)
         
-        # Update state
-        state['news_intelligence_results'] = result
-        state['current_step'] = 'news_intelligence_complete'
-        
+        updates: dict = {
+            "news_intelligence_results": result,
+            "current_step": "news_intelligence_complete",
+        }
         if not result['success']:
-            state['error'] = result.get('error', 'News intelligence failed')
-            
-        return state
+            updates["error"] = result.get('error', 'News intelligence failed')
+
+        return updates
         
     except Exception as e:
         print(f"News intelligence node error: {e}")
-        state['error'] = str(e)
-        state['current_step'] = 'error'
-        return state 
+        return {"error": str(e), "current_step": "error"}

src/agents/portfolio_manager_agent.py

@@ -389,31 +389,19 @@ def read_historical_context(symbol: str, analysis_date: Optional[str] = None) ->
         return []
 
 
-async def portfolio_manager_agent_node(state: AgentState) -> AgentState:
-    """
-    Portfolio Manager Agent node for the workflow.
-    
-    Args:
-        state: Current state of the workflow
-        
-    Returns:
-        Updated state with portfolio management results
-    """
+async def portfolio_manager_agent_node(state: AgentState) -> dict:
+    """Portfolio Manager Agent node. Returns partial state updates."""
     try:
         symbols = state.get('symbols', [])
         if not symbols:
-            state['error'] = "No symbols found in state for Portfolio Manager"
-            return state
-            
-        # Since we process one symbol at a time in this design
+            return {"error": "No symbols found in state for Portfolio Manager"}
+
         symbol = symbols[0]
         
-        # Get data directly from the state
         tech_results = state.get('technical_analysis_results', {})
         news_results = state.get('news_intelligence_results', {})
         data_collection_results = state.get('data_collection_results', {})
         
-        # Add data_collection_results to tech_results for access to basic_financials
         if isinstance(tech_results, dict):
             tech_results_with_data = {
                 **tech_results,
@@ -424,22 +412,18 @@ async def portfolio_manager_agent_node(state: AgentState) -> AgentState:
                 'data_collection_results': data_collection_results
             }
         
-        # Analyze portfolio for the single symbol
         analysis_date = state.get('analysis_date')
         analysis_result = await analyze_portfolio(
             symbol, tech_results_with_data, news_results, analysis_date
         )
         
-        # Structure the result under the symbol key
         all_results = {symbol: analysis_result}
         
-        # Update the main state with the results
-        state['portfolio_manager_results'] = all_results
-        state['current_step'] = 'portfolio_management_complete'
-        
-        return state
+        return {
+            "portfolio_manager_results": all_results,
+            "current_step": "portfolio_management_complete",
+        }
         
     except Exception as e:
         print(f"Error in portfolio_manager_agent_node: {e}")
-        state['error'] = f"Portfolio Manager Agent failed: {e}"
-        return state 
+        return {"error": f"Portfolio Manager Agent failed: {e}"}

src/agents/technical_analysis_agent.py

@@ -106,37 +106,25 @@ async def analyze_technical(symbol: str, analysis_date: Optional[str] = None, ma
         }
 
 
-async def technical_analysis_agent_node(state: AgentState) -> AgentState:
-    """
-    LangGraph node for technical analysis.
-    
-    Args:
-        state: Current workflow state
-        
-    Returns:
-        Updated state with technical analysis results
-    """
+async def technical_analysis_agent_node(state: AgentState) -> dict:
+    """LangGraph node for technical analysis. Returns partial state updates."""
     try:
-        # Get symbol, analysis_date and market data from previous agent
         symbol = state['symbols'][0] if state['symbols'] else 'AAPL'
         analysis_date = state['analysis_date']
         data_collection_results = state.get('data_collection_results')
         market_data = data_collection_results.get('market_data') if data_collection_results else None
         
-        # Perform technical analysis with analysis_date
         result = await analyze_technical(symbol, analysis_date, market_data)
         
-        # Update state
-        state['technical_analysis_results'] = result
-        state['current_step'] = 'technical_analysis_complete'
-        
+        updates: dict = {
+            "technical_analysis_results": result,
+            "current_step": "technical_analysis_complete",
+        }
         if not result['success']:
-            state['error'] = result.get('error', 'Technical analysis failed')
-            
-        return state
+            updates["error"] = result.get('error', 'Technical analysis failed')
+
+        return updates
         
     except Exception as e:
         print(f"Technical analysis node error: {e}")
-        state['error'] = str(e)
-        state['current_step'] = 'error'
-        return state 
+        return {"error": str(e), "current_step": "error"}

src/core/memory.py

@@ -1,6 +1,7 @@
 import json
 import os
 import time
+import requests
 from .logger import get_logger
 
 logger = get_logger(__name__)
@@ -8,12 +9,65 @@ logger = get_logger(__name__)
 SEEN_TICKERS_FILE = "seen_tickers.json"
 MEMORY_TTL_SECONDS = 30 * 24 * 60 * 60  # 30 days
 
+# VPS Data API (optional — falls back to local JSON if not set)
+VPS_API_URL = os.getenv("VPS_API_URL", "").rstrip("/")
+VPS_API_KEY = os.getenv("VPS_API_KEY", "")
+
+
+def _vps_headers() -> dict:
+    return {"X-API-Key": VPS_API_KEY, "Content-Type": "application/json"}
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
 
 def load_seen_tickers() -> dict[str, float]:
     """Load the seen-tickers ledger, pruning entries older than 30 days.
 
     Values are Unix timestamps (float).
+    Tries VPS API first, falls back to local JSON file.
     """
+    if VPS_API_URL:
+        try:
+            resp = requests.get(f"{VPS_API_URL}/seen-tickers", headers=_vps_headers(), timeout=5)
+            resp.raise_for_status()
+            data = resp.json()
+            logger.debug("Loaded %d seen tickers from VPS", len(data))
+            return data
+        except Exception as exc:
+            logger.warning("VPS seen-tickers unavailable, using local fallback: %s", exc)
+
+    return _load_local()
+
+
+def mark_ticker_seen(ticker: str, region: str = "USA") -> None:
+    """Record a ticker as recently analysed."""
+    if VPS_API_URL:
+        try:
+            resp = requests.post(
+                f"{VPS_API_URL}/seen-tickers",
+                headers=_vps_headers(),
+                json={"ticker": ticker, "region": region},
+                timeout=5,
+            )
+            resp.raise_for_status()
+            logger.debug("Marked %s as seen on VPS", ticker)
+            return
+        except Exception as exc:
+            logger.warning("VPS mark_ticker_seen failed, using local fallback: %s", exc)
+
+    # Local fallback
+    data = _load_local()
+    data[ticker] = time.time()
+    _save(data)
+
+
+# ---------------------------------------------------------------------------
+# Local JSON fallback (original behavior)
+# ---------------------------------------------------------------------------
+
+def _load_local() -> dict[str, float]:
     if not os.path.exists(SEEN_TICKERS_FILE):
         return {}
     try:
@@ -23,7 +77,6 @@ def load_seen_tickers() -> dict[str, float]:
         now = time.time()
         cleaned: dict[str, float] = {}
         for ticker, ts in raw.items():
-            # Support both unix timestamps and ISO strings (legacy)
             if isinstance(ts, str):
                 try:
                     from datetime import datetime, timezone
@@ -41,13 +94,6 @@ def load_seen_tickers() -> dict[str, float]:
         return {}
 
 
-def mark_ticker_seen(ticker: str) -> None:
-    """Record a ticker as recently analysed."""
-    data = load_seen_tickers()
-    data[ticker] = time.time()
-    _save(data)
-
-
 def _save(data: dict) -> None:
     try:
         with open(SEEN_TICKERS_FILE, "w") as f:

src/core/state.py

@@ -1,4 +1,5 @@
-from typing import TypedDict
+import operator
+from typing import Annotated, TypedDict
 
 
 class AgentState(TypedDict, total=False):
@@ -8,7 +9,7 @@ class AgentState(TypedDict, total=False):
     """
     region: str
     ticker: str
-    candidates: list
+    candidates: Annotated[list, operator.add]
     company_name: str
     market_cap: float
     is_small_cap: bool
@@ -19,4 +20,4 @@ class AgentState(TypedDict, total=False):
     final_report: str
     chart_data: bytes
     manual_search: bool
-    candidate_scores: list  # [{ticker, score, metrics}, ...]
+    candidate_scores: Annotated[list, operator.add]

src/models/verdict.py

@@ -0,0 +1,34 @@
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class InvestmentVerdict(BaseModel):
+    """Structured analyst verdict returned by the Senior Broker LLM."""
+
+    quantitative_base: str = Field(
+        description="Price vs calculated valuation, margin of safety math"
+    )
+    lynch_pitch: str = Field(
+        description="What insiders are doing + the one catalyst"
+    )
+    munger_invert: str = Field(
+        description="How an investor could lose money, the bear evidence"
+    )
+    verdict: Literal["STRONG BUY", "BUY", "WATCH", "AVOID"]
+    bottom_line: str = Field(
+        description="One-sentence final summary"
+    )
+
+    def to_report(self) -> str:
+        """Render as the markdown report format the pipeline expects."""
+        return (
+            f"### THE QUANTITATIVE BASE (Graham / Asset Play)\n"
+            f"{self.quantitative_base}\n\n"
+            f"### THE LYNCH PITCH (Why I would own this)\n"
+            f"{self.lynch_pitch}\n\n"
+            f"### THE MUNGER INVERT (How I could lose money)\n"
+            f"{self.munger_invert}\n\n"
+            f"### FINAL VERDICT\n"
+            f"**{self.verdict}** — {self.bottom_line}"
+        )

src/portfolio_tracker.py

@@ -1,6 +1,8 @@
 import json
 import os
 from datetime import datetime
+
+import requests
 import yfinance as yf
 from src.core.logger import get_logger
 from src.core.ticker_utils import normalize_price
@@ -9,29 +11,77 @@ logger = get_logger(__name__)
 
 PORTFOLIO_FILE = "paper_portfolio.json"
 
-
-def record_paper_trade(ticker: str, entry_price: float, verdict: str, source: str) -> None:
-    """Save a BUY/STRONG BUY/WATCH recommendation to the paper portfolio."""
-    v_upper = verdict.strip().upper()
-
-    trade_type = None
-    if "STRONG BUY" in v_upper:
-        trade_type = "STRONG BUY"
-    elif " BUY" in v_upper or v_upper.startswith("BUY"):
-        trade_type = "BUY"
-    elif "WATCH" in v_upper:
-        trade_type = "WATCH"
+# VPS Data API (optional — falls back to local JSON if not set)
+VPS_API_URL = os.getenv("VPS_API_URL", "").rstrip("/")
+VPS_API_KEY = os.getenv("VPS_API_KEY", "")
+
+
+def _vps_headers() -> dict:
+    return {"X-API-Key": VPS_API_KEY, "Content-Type": "application/json"}
+
+
+def record_paper_trade(
+    ticker: str,
+    entry_price: float,
+    verdict: str,
+    source: str,
+    structured_verdict: str | None = None,
+) -> None:
+    """Save a BUY/STRONG BUY/WATCH recommendation to the paper portfolio.
+
+    When *structured_verdict* is supplied (from ``InvestmentVerdict.verdict``),
+    it is used directly, skipping brittle string matching on the full report.
+    """
+    if structured_verdict:
+        _VALID = {"STRONG BUY", "BUY", "WATCH"}
+        trade_type = structured_verdict if structured_verdict in _VALID else None
+    else:
+        v_upper = verdict.strip().upper()
+        trade_type = None
+        if "STRONG BUY" in v_upper:
+            trade_type = "STRONG BUY"
+        elif " BUY" in v_upper or v_upper.startswith("BUY"):
+            trade_type = "BUY"
+        elif "WATCH" in v_upper:
+            trade_type = "WATCH"
 
     if not trade_type:
         return
 
+    today = datetime.now().strftime("%Y-%m-%d")
+
+    # Try VPS first
+    if VPS_API_URL:
+        try:
+            resp = requests.post(
+                f"{VPS_API_URL}/portfolio",
+                headers=_vps_headers(),
+                json={
+                    "ticker": ticker,
+                    "entry_price": entry_price,
+                    "date": today,
+                    "verdict": trade_type,
+                    "source": source,
+                },
+                timeout=5,
+            )
+            resp.raise_for_status()
+            result = resp.json()
+            if result.get("status") == "duplicate":
+                logger.info("Duplicate trade skipped (VPS): %s on %s", ticker, today)
+            else:
+                logger.info("Paper trade recorded (VPS): %s %s @ $%.2f", trade_type, ticker, entry_price)
+            return
+        except Exception as exc:
+            logger.warning("VPS record_paper_trade failed, using local fallback: %s", exc)
+
+    # Local fallback
     try:
         portfolio = []
         if os.path.exists(PORTFOLIO_FILE):
             with open(PORTFOLIO_FILE, "r") as f:
                 portfolio = json.load(f)
 
-        today = datetime.now().strftime("%Y-%m-%d")
         for trade in portfolio:
             if trade["ticker"] == ticker and trade["date"] == today:
                 logger.info("Duplicate trade skipped: %s on %s", ticker, today)
@@ -58,6 +108,54 @@ def record_paper_trade(ticker: str, entry_price: float, verdict: str, source: st
 
 def evaluate_portfolio() -> str:
     """Read the paper portfolio and calculate live performance."""
+
+    # Try VPS first
+    if VPS_API_URL:
+        try:
+            resp = requests.get(
+                f"{VPS_API_URL}/portfolio/evaluate",
+                headers=_vps_headers(),
+                timeout=30,
+            )
+            resp.raise_for_status()
+            data = resp.json()
+
+            if not data.get("trades"):
+                return "**Paper Portfolio is empty.** The Agent hasn't tracked any stocks yet."
+
+            report = "## PrimoGreedy Agent Track Record\n\n"
+            report += "| Ticker | Date Called | Entry Price | Current Price | Return | Verdict |\n"
+            report += "|--------|-------------|-------------|---------------|--------|--------|\n"
+
+            for t in data["trades"]:
+                if t["gain_pct"] is not None:
+                    emoji = "+" if t["gain_pct"] > 0 else ""
+                    report += (
+                        f"| **{t['ticker']}** | {t['date']} | ${t['entry']:.2f} | "
+                        f"${t['current']:.2f} | {emoji}{t['gain_pct']:.2f}% | {t['verdict']} |\n"
+                    )
+                else:
+                    report += (
+                        f"| **{t['ticker']}** | {t['date']} | ${t['entry']:.2f} | "
+                        f"Error | N/A | {t['verdict']} |\n"
+                    )
+
+            report += f"\n### Agent Performance Summary\n"
+            report += f"- **Total Calls:** {data['total_calls']}\n"
+            report += f"- **Win Rate:** {data['win_rate']}%\n"
+            report += f"- **Average Return per Trade:** {data['avg_return']}%\n"
+
+            return report
+
+        except Exception as exc:
+            logger.warning("VPS evaluate_portfolio failed, using local fallback: %s", exc)
+
+    # Local fallback (original behavior)
+    return _evaluate_local()
+
+
+def _evaluate_local() -> str:
+    """Evaluate portfolio from local JSON file."""
     if not os.path.exists(PORTFOLIO_FILE):
         return "**Paper Portfolio is empty.** The Agent hasn't tracked any stocks yet."
 
@@ -74,7 +172,7 @@ def evaluate_portfolio() -> str:
 
         report = "## PrimoGreedy Agent Track Record\n\n"
         report += "| Ticker | Date Called | Entry Price | Current Price | Return | Verdict |\n"
-        report += "|--------|-------------|-------------|---------------|--------|---------|\n"
+        report += "|--------|-------------|-------------|---------------|--------|--------|\n"
 
         for trade in portfolio:
             ticker = trade["ticker"]

src/whale_hunter.py

@@ -1,8 +1,11 @@
 """Daily automated micro-cap hunter (GitHub Actions cron).
 
-Pipeline:  Scout -> Gatekeeper -> Analyst -> Email
+Pipeline per region:  Scout -> Gatekeeper -> Analyst -> Email
 
-The Scout now uses a two-pronged discovery approach:
+An outer orchestrator graph dispatches all regions in parallel via the
+LangGraph ``Send`` API, then collects results.
+
+The Scout uses a two-pronged discovery approach:
   1. yFinance screener for systematic micro-cap filtering
   2. Brave Search for trending/momentum signals
   3. Quantitative scoring to pick the best candidate
@@ -11,10 +14,14 @@ Both feeds are merged, scored, and only the top candidate proceeds to
 the expensive LLM analyst step.
 """
 
+import operator
 import os
 import signal
 import time
-from langgraph.graph import StateGraph, END
+from typing import Annotated, Literal, TypedDict
+from langgraph.graph import StateGraph, START, END
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.types import Command, RetryPolicy, Send
 
 from src.llm import get_llm, invoke_with_fallback
 from src.finance_tools import (
@@ -110,16 +117,22 @@ def scout_node(state):
     return {"ticker": ticker, "candidates": rest}
 
 
-def gatekeeper_node(state):
-    """Validate the candidate against hard financial criteria."""
+def gatekeeper_node(state) -> Command[Literal["analyst", "scout", "email"]]:
+    """Validate the candidate against hard financial criteria. Routes via Command."""
     import yfinance as yf
 
     ticker = state.get("ticker", "NONE")
     retries = state.get("retry_count", 0)
 
+    def _fail_route(new_retries: int) -> str:
+        if new_retries > MAX_RETRIES:
+            return "email"
+        return "scout"
+
     if ticker == "NONE":
         logger.warning("No ticker to evaluate")
-        return {"is_small_cap": False, "retry_count": retries + 1}
+        update = {"is_small_cap": False, "retry_count": retries + 1}
+        return Command(update=update, goto=_fail_route(retries + 1))
 
     logger.info("Gatekeeper evaluating %s...", ticker)
     try:
@@ -135,16 +148,19 @@ def gatekeeper_node(state):
 
         if price > MAX_PRICE_PER_SHARE:
             logger.info("%s rejected — price $%.2f > $%.2f", ticker, price, MAX_PRICE_PER_SHARE)
-            return {"is_small_cap": False, "retry_count": retries + 1}
+            update = {"is_small_cap": False, "retry_count": retries + 1}
+            return Command(update=update, goto=_fail_route(retries + 1))
 
         if not (MIN_MARKET_CAP < mkt_cap < MAX_MARKET_CAP):
             logger.info("%s rejected — cap $%s out of range", ticker, f"{mkt_cap:,.0f}")
-            return {"is_small_cap": False, "retry_count": retries + 1}
+            update = {"is_small_cap": False, "retry_count": retries + 1}
+            return Command(update=update, goto=_fail_route(retries + 1))
 
         health = check_financial_health(ticker, info)
         if health["status"] == "FAIL":
             logger.info("%s rejected — %s", ticker, health["reason"])
-            return {"is_small_cap": False, "retry_count": retries + 1}
+            update = {"is_small_cap": False, "retry_count": retries + 1}
+            return Command(update=update, goto=_fail_route(retries + 1))
 
         sector = health["metrics"].get("sector", "N/A")
         logger.info(
@@ -152,16 +168,18 @@ def gatekeeper_node(state):
             ticker, price, f"{mkt_cap:,.0f}", sector,
         )
 
-        return {
+        update = {
             "market_cap": mkt_cap,
             "is_small_cap": True,
             "company_name": name,
             "financial_data": info,
         }
+        return Command(update=update, goto="analyst")
 
     except Exception as exc:
         logger.error("yFinance error for %s: %s", ticker, exc)
-        return {"is_small_cap": False, "retry_count": retries + 1}
+        update = {"is_small_cap": False, "retry_count": retries + 1}
+        return Command(update=update, goto=_fail_route(retries + 1))
 
 
 def analyst_node(state):
@@ -237,11 +255,20 @@ def analyst_node(state):
     """
 
     try:
-        verdict = invoke_with_fallback(prompt)
-        record_paper_trade(ticker, price, verdict, source="Morning Cron")
+        from src.models.verdict import InvestmentVerdict
+        structured_llm = get_llm().with_structured_output(InvestmentVerdict)
+        result = structured_llm.invoke(prompt)
+        verdict = result.to_report()
+        record_paper_trade(ticker, price, verdict, source="Morning Cron",
+                           structured_verdict=result.verdict)
     except Exception as exc:
-        logger.error("LLM analysis failed for %s: %s", ticker, exc)
-        verdict = f"LLM analysis unavailable: {exc}"
+        logger.warning("Structured output failed for %s, falling back to plain LLM: %s", ticker, exc)
+        try:
+            verdict = invoke_with_fallback(prompt)
+            record_paper_trade(ticker, price, verdict, source="Morning Cron")
+        except Exception as exc2:
+            logger.error("LLM analysis failed for %s: %s", ticker, exc2)
+            verdict = f"LLM analysis unavailable: {exc2}"
 
     return {"final_verdict": verdict}
 
@@ -281,35 +308,71 @@ def email_node(state):
     return {}
 
 
-# --- GRAPH ---
+# ---------------------------------------------------------------------------
+# Per-region subgraph  (scout -> gatekeeper -> analyst -> email)
+# ---------------------------------------------------------------------------
 
-workflow = StateGraph(AgentState)
-workflow.add_node("scout", scout_node)
-workflow.add_node("gatekeeper", gatekeeper_node)
-workflow.add_node("analyst", analyst_node)
-workflow.add_node("email", email_node)
-workflow.set_entry_point("scout")
+_api_retry = RetryPolicy(max_attempts=3, initial_interval=2.0)
 
+_region_workflow = StateGraph(AgentState)
+_region_workflow.add_node("scout", scout_node, retry=_api_retry)
+_region_workflow.add_node("gatekeeper", gatekeeper_node, retry=_api_retry)
+_region_workflow.add_node("analyst", analyst_node, retry=_api_retry)
+_region_workflow.add_node("email", email_node, retry=_api_retry)
+_region_workflow.add_edge(START, "scout")
+_region_workflow.add_edge("scout", "gatekeeper")
+_region_workflow.add_edge("analyst", "email")
+_region_workflow.add_edge("email", END)
+_region_app = _region_workflow.compile(checkpointer=InMemorySaver())
 
-def check_status(state):
-    if state.get("is_small_cap"):
-        return "analyst"
-    if state.get("retry_count", 0) > MAX_RETRIES:
-        return "email"
-    return "scout"
 
+# ---------------------------------------------------------------------------
+# Orchestrator graph  (parallel fan-out via Send API)
+# ---------------------------------------------------------------------------
+
+class GlobalHunterState(TypedDict, total=False):
+    regions: list[str]
+    region_results: Annotated[list, operator.add]
+
+
+def dispatch_regions(state: GlobalHunterState):
+    """Fan-out: emit one Send per region so they run in parallel."""
+    return [
+        Send("hunt_region", {"region": r})
+        for r in state["regions"]
+    ]
+
+
+def hunt_region(state) -> dict:
+    """Invoke the full per-region pipeline and report back."""
+    region = state.get("region", "USA")
+    logger.info("--- Initiating hunt for %s ---", region)
+    try:
+        config = {
+            "configurable": {"thread_id": f"hunt-{region.lower()}"},
+            "recursion_limit": 30,
+        }
+        _region_app.invoke(
+            {"region": region, "retry_count": 0, "ticker": ""},
+            config,
+        )
+        logger.info("%s hunt complete.", region)
+        return {"region_results": [{"region": region, "success": True}]}
+    except Exception as exc:
+        logger.error("Error in %s: %s", region, exc, exc_info=True)
+        return {"region_results": [{"region": region, "success": False, "error": str(exc)}]}
 
-workflow.add_edge("scout", "gatekeeper")
-workflow.add_conditional_edges(
-    "gatekeeper",
-    check_status,
-    {"analyst": "analyst", "scout": "scout", "email": "email"},
-)
-workflow.add_edge("analyst", "email")
-workflow.add_edge("email", END)
-app = workflow.compile()
 
-# --- EXECUTION ---
+_orchestrator = StateGraph(GlobalHunterState)
+_orchestrator.add_node("hunt_region", hunt_region, retry=_api_retry)
+_orchestrator.add_conditional_edges(START, dispatch_regions, ["hunt_region"])
+_orchestrator.add_edge("hunt_region", END)
+app = _orchestrator.compile(checkpointer=InMemorySaver())
+
+
+# ---------------------------------------------------------------------------
+# Execution
+# ---------------------------------------------------------------------------
 
 if __name__ == "__main__":
     logger.info("Starting Global Micro-Cap Hunter (Screener + Brave Edition)...")
@@ -323,13 +386,11 @@ if __name__ == "__main__":
 
     regions = ["USA", "UK", "Canada", "Australia"]
 
-    for market in regions:
-        logger.info("--- Initiating hunt for %s ---", market)
-        try:
-            app.invoke({"region": market, "retry_count": 0, "ticker": ""})
-            logger.info("%s hunt complete.", market)
-            time.sleep(5)
-        except Exception as exc:
-            logger.error("Error in %s: %s", market, exc, exc_info=True)
+    config = {"configurable": {"thread_id": "global-hunt"}, "recursion_limit": 30}
+    result = app.invoke({"regions": regions}, config)
+
+    for entry in result.get("region_results", []):
+        status = "OK" if entry.get("success") else f"FAILED: {entry.get('error', 'unknown')}"
+        logger.info("Region %s: %s", entry.get("region"), status)
 
     logger.info("Global mission complete.")

src/workflows/workflow.py

@@ -1,5 +1,7 @@
 from typing import Dict, Any, Optional
-from langgraph.graph import StateGraph, END
+from langgraph.graph import StateGraph, START, END
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.types import RetryPolicy
 from .state import AgentState, create_initial_state
 from ..agents.data_collection_agent import data_collection_agent_node
 from ..agents.technical_analysis_agent import technical_analysis_agent_node
@@ -7,72 +9,63 @@ from ..agents.news_intelligence_agent import news_intelligence_agent_node
 from ..agents.portfolio_manager_agent import portfolio_manager_agent_node
 
 
-def debug_state(state: AgentState, agent_name: str) -> AgentState:
-    """Debug function to log state after each agent."""
+def _log_partial(updates: dict, agent_name: str) -> None:
+    """Log interesting fields from a partial-state update dict."""
     print(f"\n{agent_name} Agent Complete:")
-    
-    # Basic info
-    analysis_date = state.get('analysis_date', 'N/A')
-    symbol = state['symbols'][0] if state.get('symbols') else 'N/A'
-    print(f"Date: {analysis_date} | Symbol: {symbol}")
-    
-    # Data Collection Results
-    data_results = state.get('data_collection_results')
-    if data_results and agent_name == "Data Collection":
-        market_data = data_results.get('market_data', {})
-        current_price = market_data.get('current_price', 'N/A')
-        print(f"Current Price: ${current_price}")
-    
-    # Technical Analysis Results
-    tech_results = state.get('technical_analysis_results')
-    if tech_results and agent_name == "Technical Analysis":
-        success = tech_results.get('success', False)
-        print(f"Technical Success: {success}")
-    
-    # News Intelligence Results
-    news_results = state.get('news_intelligence_results')
-    if news_results and agent_name == "News Intelligence":
-        success = news_results.get('success', False)
-        print(f"News Success: {success}")
-    
-    # Portfolio Manager Results
-    portfolio_results = state.get('portfolio_manager_results')
-    if portfolio_results and agent_name == "Portfolio Manager":
-        symbol_data = portfolio_results.get(symbol, {})
-        if symbol_data and symbol_data.get('success'):
-            signal = symbol_data.get('trading_signal', 'N/A')
-            confidence = symbol_data.get('confidence_level', 'N/A')
-            print(f"Signal: {signal} | Confidence: {confidence}")
-    
-    # Error state
-    if state.get('error'):
-        print(f"Error: {state.get('error')}")
-    
-    return state
 
+    if agent_name == "Data Collection":
+        data_results = updates.get('data_collection_results')
+        if data_results:
+            market_data = data_results.get('market_data', {})
+            print(f"Current Price: ${market_data.get('current_price', 'N/A')}")
+
+    elif agent_name == "Technical Analysis":
+        tech_results = updates.get('technical_analysis_results')
+        if tech_results:
+            print(f"Technical Success: {tech_results.get('success', False)}")
+
+    elif agent_name == "News Intelligence":
+        news_results = updates.get('news_intelligence_results')
+        if news_results:
+            print(f"News Success: {news_results.get('success', False)}")
+
+    elif agent_name == "Portfolio Manager":
+        portfolio_results = updates.get('portfolio_manager_results', {})
+        for sym, sym_data in portfolio_results.items():
+            if sym_data and sym_data.get('success'):
+                print(f"Signal: {sym_data.get('trading_signal', 'N/A')} | "
+                      f"Confidence: {sym_data.get('confidence_level', 'N/A')}")
 
-async def debug_data_collection_node(state: AgentState) -> AgentState:
+    if updates.get('error'):
+        print(f"Error: {updates['error']}")
+
+
+async def debug_data_collection_node(state: AgentState) -> dict:
     """Data collection node with debug output."""
-    result = await data_collection_agent_node(state)
-    return debug_state(result, "Data Collection")
+    updates = await data_collection_agent_node(state)
+    _log_partial(updates, "Data Collection")
+    return updates
 
 
-async def debug_technical_analysis_node(state: AgentState) -> AgentState:
-    """Technical analysis node with debug output.""" 
-    result = await technical_analysis_agent_node(state)
-    return debug_state(result, "Technical Analysis")
+async def debug_technical_analysis_node(state: AgentState) -> dict:
+    """Technical analysis node with debug output."""
+    updates = await technical_analysis_agent_node(state)
+    _log_partial(updates, "Technical Analysis")
+    return updates
 
 
-async def debug_news_intelligence_node(state: AgentState) -> AgentState:
+async def debug_news_intelligence_node(state: AgentState) -> dict:
     """News intelligence node with debug output."""
-    result = await news_intelligence_agent_node(state)  
-    return debug_state(result, "News Intelligence")
+    updates = await news_intelligence_agent_node(state)
+    _log_partial(updates, "News Intelligence")
+    return updates
 
 
-async def debug_portfolio_manager_node(state: AgentState) -> AgentState:
+async def debug_portfolio_manager_node(state: AgentState) -> dict:
     """Portfolio manager node with debug output."""
-    result = await portfolio_manager_agent_node(state)
-    return debug_state(result, "Portfolio Manager")
+    updates = await portfolio_manager_agent_node(state)
+    _log_partial(updates, "Portfolio Manager")
+    return updates
 
 
 def create_workflow() -> StateGraph:
@@ -83,16 +76,17 @@ def create_workflow() -> StateGraph:
         StateGraph: Configured workflow graph
     """
     # Initialize workflow
+    _api_retry = RetryPolicy(max_attempts=3, initial_interval=2.0)
+
     workflow = StateGraph(AgentState)
     
-    # Add nodes with debug output
-    workflow.add_node("data_collection", debug_data_collection_node)
-    workflow.add_node("technical_analysis", debug_technical_analysis_node)
-    workflow.add_node("news_intelligence", debug_news_intelligence_node)
-    workflow.add_node("portfolio_manager", debug_portfolio_manager_node)
+    workflow.add_node("data_collection", debug_data_collection_node, retry=_api_retry)
+    workflow.add_node("technical_analysis", debug_technical_analysis_node, retry=_api_retry)
+    workflow.add_node("news_intelligence", debug_news_intelligence_node, retry=_api_retry)
+    workflow.add_node("portfolio_manager", debug_portfolio_manager_node, retry=_api_retry)
     
     # Define linear flow
-    workflow.set_entry_point("data_collection")
+    workflow.add_edge(START, "data_collection")
     workflow.add_edge("data_collection", "technical_analysis")
     workflow.add_edge("technical_analysis", "news_intelligence")
     workflow.add_edge("news_intelligence", "portfolio_manager")
@@ -116,13 +110,14 @@ async def run_analysis(symbols: list[str], session_id: str = "default", analysis
     try:
         # Create workflow
         workflow = create_workflow()
-        app = workflow.compile()
+        app = workflow.compile(checkpointer=InMemorySaver())
         
         # Initialize state with analysis date
         initial_state = create_initial_state(session_id, symbols, analysis_date)
         
         # Run workflow
-        result = await app.ainvoke(initial_state)
+        config = {"configurable": {"thread_id": session_id}, "recursion_limit": 30}
+        result = await app.ainvoke(initial_state, config)
         
         # Extract results
         return {

vps/api.py

@@ -0,0 +1,262 @@
+"""PrimoGreedy Data API — FastAPI wrapper around DuckDB.
+
+Serves seen tickers, paper portfolio, and agent run logs over HTTP.
+Secured with X-API-Key header. Runs behind Tailscale (private network).
+
+Usage:
+    uvicorn api:app --host 0.0.0.0 --port 8000
+"""
+
+import os
+import time
+from contextlib import asynccontextmanager
+from datetime import datetime, timezone
+from typing import Optional
+
+import duckdb
+import yfinance as yf
+from dotenv import load_dotenv
+from fastapi import FastAPI, Header, HTTPException, Query
+from pydantic import BaseModel
+
+load_dotenv()
+
+DB_PATH = os.getenv("DUCKDB_PATH", "/home/ubuntu/primogreedy/data.duckdb")
+API_KEY = os.getenv("VPS_API_KEY", "5ZhJ_T2gTTQp-LAJKdWMvKJgQSqFU8MSfFDAi04tNr0")
+MEMORY_TTL_SECONDS = 30 * 24 * 60 * 60  # 30 days
+
+
+# ---------------------------------------------------------------------------
+# Database helpers
+# ---------------------------------------------------------------------------
+
+def get_db() -> duckdb.DuckDBPyConnection:
+    """Return a fresh connection (DuckDB is single-writer, but reads are fine)."""
+    return duckdb.connect(DB_PATH)
+
+
+def init_db():
+    """Create tables if they don't exist."""
+    con = get_db()
+    con.execute("""
+        CREATE SEQUENCE IF NOT EXISTS portfolio_seq START 1;
+
+        CREATE TABLE IF NOT EXISTS seen_tickers (
+            ticker    VARCHAR NOT NULL PRIMARY KEY,
+            region    VARCHAR DEFAULT 'USA',
+            seen_at   TIMESTAMP DEFAULT current_timestamp
+        );
+
+        CREATE TABLE IF NOT EXISTS paper_portfolio (
+            id          INTEGER PRIMARY KEY DEFAULT nextval('portfolio_seq'),
+            ticker      VARCHAR NOT NULL,
+            entry_price DOUBLE NOT NULL,
+            date        DATE NOT NULL,
+            verdict     VARCHAR NOT NULL,
+            source      VARCHAR DEFAULT 'unknown',
+            created_at  TIMESTAMP DEFAULT current_timestamp,
+            UNIQUE (ticker, date)
+        );
+
+        CREATE TABLE IF NOT EXISTS agent_runs (
+            id          VARCHAR PRIMARY KEY,
+            ticker      VARCHAR NOT NULL,
+            timestamp   TIMESTAMP DEFAULT current_timestamp,
+            status      VARCHAR NOT NULL,
+            model       VARCHAR,
+            latency_ms  INTEGER,
+            region      VARCHAR DEFAULT 'USA'
+        );
+    """)
+    con.close()
+
+
+# ---------------------------------------------------------------------------
+# App lifecycle
+# ---------------------------------------------------------------------------
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
+    init_db()
+    yield
+
+
+app = FastAPI(
+    title="PrimoGreedy Data API",
+    version="1.0.0",
+    lifespan=lifespan,
+)
+
+
+# ---------------------------------------------------------------------------
+# Auth
+# ---------------------------------------------------------------------------
+
+def verify_key(x_api_key: str = Header(...)):
+    if x_api_key != API_KEY:
+        raise HTTPException(status_code=401, detail="Invalid API key")
+
+
+# ---------------------------------------------------------------------------
+# Pydantic models
+# ---------------------------------------------------------------------------
+
+class SeenTickerIn(BaseModel):
+    ticker: str
+    region: str = "USA"
+
+
+class TradeIn(BaseModel):
+    ticker: str
+    entry_price: float
+    date: str  # YYYY-MM-DD
+    verdict: str
+    source: str = "unknown"
+
+
+# ---------------------------------------------------------------------------
+# Health
+# ---------------------------------------------------------------------------
+
+@app.get("/health")
+def health():
+    return {"status": "ok", "db": DB_PATH}
+
+
+# ---------------------------------------------------------------------------
+# Seen Tickers
+# ---------------------------------------------------------------------------
+
+@app.get("/seen-tickers")
+def get_seen_tickers(x_api_key: str = Header(...)):
+    verify_key(x_api_key)
+    con = get_db()
+    cutoff = time.time() - MEMORY_TTL_SECONDS
+    cutoff_ts = datetime.fromtimestamp(cutoff, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+
+    rows = con.execute(
+        "SELECT ticker, epoch(seen_at) as ts FROM seen_tickers WHERE seen_at >= ?",
+        [cutoff_ts],
+    ).fetchall()
+    con.close()
+
+    return {row[0]: row[1] for row in rows}
+
+
+@app.post("/seen-tickers")
+def mark_seen(body: SeenTickerIn, x_api_key: str = Header(...)):
+    verify_key(x_api_key)
+    con = get_db()
+    con.execute(
+        """INSERT INTO seen_tickers (ticker, region, seen_at)
+           VALUES (?, ?, now())
+           ON CONFLICT (ticker) DO UPDATE SET seen_at = now(), region = ?""",
+        [body.ticker, body.region, body.region],
+    )
+    con.close()
+    return {"status": "ok", "ticker": body.ticker}
+
+
+# ---------------------------------------------------------------------------
+# Portfolio
+# ---------------------------------------------------------------------------
+
+@app.get("/portfolio")
+def get_portfolio(x_api_key: str = Header(...)):
+    verify_key(x_api_key)
+    con = get_db()
+    rows = con.execute(
+        "SELECT ticker, entry_price, date, verdict, source FROM paper_portfolio ORDER BY date DESC"
+    ).fetchall()
+    con.close()
+
+    return [
+        {
+            "ticker": r[0],
+            "entry_price": r[1],
+            "date": str(r[2]),
+            "verdict": r[3],
+            "source": r[4],
+        }
+        for r in rows
+    ]
+
+
+@app.post("/portfolio")
+def record_trade(body: TradeIn, x_api_key: str = Header(...)):
+    verify_key(x_api_key)
+    con = get_db()
+    try:
+        con.execute(
+            """INSERT INTO paper_portfolio (ticker, entry_price, date, verdict, source)
+               VALUES (?, ?, ?, ?, ?)""",
+            [body.ticker, body.entry_price, body.date, body.verdict, body.source],
+        )
+    except duckdb.ConstraintException:
+        con.close()
+        return {"status": "duplicate", "ticker": body.ticker, "date": body.date}
+    con.close()
+    return {"status": "ok", "ticker": body.ticker}
+
+
+@app.get("/portfolio/evaluate")
+def evaluate_portfolio(x_api_key: str = Header(...)):
+    """Fetch live prices and compute P&L for all portfolio entries."""
+    verify_key(x_api_key)
+    con = get_db()
+    rows = con.execute(
+        "SELECT ticker, entry_price, date, verdict, source FROM paper_portfolio ORDER BY date"
+    ).fetchall()
+    con.close()
+
+    if not rows:
+        return {"report": "Paper Portfolio is empty.", "trades": []}
+
+    trades = []
+    total_roi = 0.0
+    winners = 0
+    valid = 0
+
+    for r in rows:
+        ticker, entry, date, verdict, source = r[0], r[1], str(r[2]), r[3], r[4]
+        try:
+            info = yf.Ticker(ticker).info
+            price = info.get("currentPrice", 0) or info.get("regularMarketPrice", 0) or 0
+            currency = info.get("currency", "USD")
+
+            # Pence → Pounds
+            if ticker.endswith(".L") or currency in ("GBp", "GBX"):
+                price = price / 100
+
+            if price > 0 and entry > 0:
+                gain = ((price - entry) / entry) * 100
+                if gain > 0:
+                    winners += 1
+                total_roi += gain
+                valid += 1
+                trades.append({
+                    "ticker": ticker, "date": date, "entry": entry,
+                    "current": round(price, 2), "gain_pct": round(gain, 2),
+                    "verdict": verdict,
+                })
+            else:
+                trades.append({
+                    "ticker": ticker, "date": date, "entry": entry,
+                    "current": None, "gain_pct": None, "verdict": verdict,
+                })
+        except Exception:
+            trades.append({
+                "ticker": ticker, "date": date, "entry": entry,
+                "current": None, "gain_pct": None, "verdict": verdict,
+            })
+
+    avg_roi = total_roi / valid if valid else 0
+    win_rate = (winners / valid * 100) if valid else 0
+
+    return {
+        "total_calls": len(rows),
+        "win_rate": round(win_rate, 1),
+        "avg_return": round(avg_roi, 2),
+        "trades": trades,
+    }

vps/deploy.sh

@@ -0,0 +1,49 @@
+#!/bin/bash
+# Deploy PrimoGreedy Data API to VPS
+# Usage: bash vps/deploy.sh
+
+set -e
+
+VPS="ubuntu@100.110.105.118"
+REMOTE_DIR="/home/ubuntu/primogreedy"
+
+echo "=== PrimoGreedy VPS Deploy ==="
+
+echo "1. Creating remote directory..."
+ssh $VPS "mkdir -p $REMOTE_DIR"
+
+echo "2. Copying files..."
+scp vps/api.py vps/requirements.txt vps/.env vps/schema.sql $VPS:$REMOTE_DIR/
+
+echo "3. Installing Python dependencies..."
+ssh $VPS "cd $REMOTE_DIR && pip3 install --break-system-packages -r requirements.txt"
+
+echo "4. Creating systemd service..."
+ssh $VPS "sudo tee /etc/systemd/system/primogreedy-api.service > /dev/null << 'EOF'
+[Unit]
+Description=PrimoGreedy Data API
+After=network.target
+
+[Service]
+Type=simple
+User=ubuntu
+WorkingDirectory=/home/ubuntu/primogreedy
+ExecStart=/usr/bin/python3 -m uvicorn api:app --host 0.0.0.0 --port 8080
+Restart=always
+RestartSec=5
+EnvironmentFile=/home/ubuntu/primogreedy/.env
+
+[Install]
+WantedBy=multi-user.target
+EOF"
+
+echo "5. Starting service..."
+ssh $VPS "sudo systemctl daemon-reload && sudo systemctl enable primogreedy-api && sudo systemctl restart primogreedy-api"
+
+echo "6. Checking status..."
+sleep 2
+ssh $VPS "sudo systemctl status primogreedy-api --no-pager -l" || true
+
+echo ""
+echo "=== Deploy complete ==="
+echo "Health check: curl http://100.110.105.118:8000/health"

vps/requirements.txt

@@ -0,0 +1,5 @@
+fastapi>=0.115.0
+uvicorn>=0.34.0
+duckdb>=1.2.0
+yfinance>=0.2.50
+python-dotenv>=1.0.0

vps/schema.sql

@@ -0,0 +1,35 @@
+-- DuckDB schema for PrimoGreedy data layer
+-- Run: duckdb data.duckdb < schema.sql
+
+-- Seen tickers — replaces seen_tickers.json
+CREATE TABLE IF NOT EXISTS seen_tickers (
+    ticker    VARCHAR NOT NULL,
+    region    VARCHAR DEFAULT 'USA',
+    seen_at   TIMESTAMP DEFAULT current_timestamp,
+    PRIMARY KEY (ticker)
+);
+
+-- Paper portfolio — replaces paper_portfolio.json
+CREATE TABLE IF NOT EXISTS paper_portfolio (
+    id          INTEGER PRIMARY KEY DEFAULT nextval('portfolio_seq'),
+    ticker      VARCHAR NOT NULL,
+    entry_price DOUBLE NOT NULL,
+    date        DATE NOT NULL,
+    verdict     VARCHAR NOT NULL,
+    source      VARCHAR DEFAULT 'unknown',
+    created_at  TIMESTAMP DEFAULT current_timestamp,
+    UNIQUE (ticker, date)  -- prevent duplicate same-day entries
+);
+
+CREATE SEQUENCE IF NOT EXISTS portfolio_seq START 1;
+
+-- Agent run log — operational metrics for LangSmith correlation
+CREATE TABLE IF NOT EXISTS agent_runs (
+    id          VARCHAR PRIMARY KEY,
+    ticker      VARCHAR NOT NULL,
+    timestamp   TIMESTAMP DEFAULT current_timestamp,
+    status      VARCHAR NOT NULL,  -- PASS / FAIL
+    model       VARCHAR,
+    latency_ms  INTEGER,
+    region      VARCHAR DEFAULT 'USA'
+);