Upload 15 files

app.py

@@ -90,6 +90,13 @@ table{font-size:13.5px!important;}
 thead th{background:var(--s2-gray-75)!important;font-weight:700!important;}
 
 .s2-footnote{color:#6e6e6e;font-size:12.5px;}
+
+/* AI output panel — big, framed, unmissable */
+.ai-panel{background:#fff;border:1.5px solid var(--s2-accent)!important;
+  border-left:6px solid var(--s2-accent)!important;border-radius:14px!important;
+  padding:18px 22px!important;min-height:240px;max-height:560px;overflow-y:auto;
+  font-size:15px;line-height:1.55;box-shadow:0 2px 10px rgba(2,101,220,.08);}
+.ai-panel:empty::after{content:"AI output will appear here";color:#9a9a9a;}
 #detail-log textarea{font-family:'Source Code Pro',monospace!important;font-size:12.5px!important;}
 """
 
@@ -128,14 +135,17 @@ def ui_explain_detail(ticker):
     # keep only the ruling chain — the per-signal diagnostics block is huge and
     # would cost a minute of silent CPU prompt-processing for no benefit
     core = txt.split("日线买卖点逐项诊断")[0][:1500]
-    prompt = ("Below is a Chan-theory (缠论) multi-timeframe decision log in Chinese "
-              "for a US stock. Do NOT repeat or quote the log. Write a fresh plain-"
-              "English explanation for a trader: 1) the final action; 2) why each "
-              "timeframe gate passed/failed; 3) what price/event would invalidate "
-              "the call. ≤160 words.\n\n" + core)
+    prompt = ("Below is a Chan-theory multi-timeframe decision log in Chinese for a "
+              "US stock. Respond in ENGLISH ONLY — translate every Chinese term; no "
+              "Chinese characters may appear in your answer. Do NOT quote the log. "
+              "Give a SHORT summary in exactly this format:\n"
+              "**Action:** <BUY/SELL/HOLD/WAIT + one clause>\n"
+              "**Why:** <2-3 short bullets: which timeframe gates passed/failed>\n"
+              "**Invalidation:** <one line: what price/event flips the call>\n"
+              "Max 80 words total.\n\n" + core)
     yield ("🤖 _Translator sub-agent (Qwen3-1.7B · llama.cpp) is reading the "
            "decision log — first words in ~5-15s…_")
-    for acc in llm_local.chat_stream(prompt, max_tokens=340, worker="fast"):
+    for acc in llm_local.chat_stream(prompt, max_tokens=220, temperature=0.1, worker="translator"):
         yield "🤖 **Translator sub-agent (Qwen3-1.7B · llama.cpp):**\n\n" + acc
 
 
@@ -162,7 +172,7 @@ def ui_rotation_ai():
               "No disclaimers.\n\nDATA:\n" + brief[:2200])
     yield ("🤖 _Narrator sub-agent (Qwen3-1.7B · llama.cpp) is reading the flow "
            "tables — first words in ~5-15s…_")
-    for acc in llm_local.chat_stream(prompt, max_tokens=340, worker="fast"):
+    for acc in llm_local.chat_stream(prompt, max_tokens=340, worker="narrator"):
         yield "🤖 **Narrator sub-agent (Qwen3-1.7B · llama.cpp):**\n\n" + acc
 
 
@@ -227,8 +237,8 @@ with gr.Blocks(title="Chan Compass · US", **_style_kw) as demo:
         with gr.Row():
             detail_pick = gr.Dropdown(choices=[], label="Ticker", scale=2)
             explain_btn = gr.Button("🌐 Explain in English (local LLM)", scale=1)
-        detail_box = gr.Textbox(lines=18, label="Ruling chain", elem_id="detail-log")
-        explain_box = gr.Markdown()
+        detail_box = gr.Textbox(lines=14, label="Ruling chain", elem_id="detail-log")
+        explain_box = gr.Markdown(elem_classes=["ai-panel"])
 
     with gr.Tab("🔄 Sector Rotation"):
         gr.Markdown("Capital rotation across the 11 SPDR sector ETFs (full S&P 500 coverage). "
@@ -243,7 +253,7 @@ with gr.Blocks(title="Chan Compass · US", **_style_kw) as demo:
         with gr.Row():
             rot_5d = gr.Dataframe(label="5-Day (week trend)", interactive=False)
             rot_20d = gr.Dataframe(label="20-Day (month trend)", interactive=False)
-        rot_ai = gr.Markdown(label="AI rotation narrative")
+        rot_ai = gr.Markdown(label="AI rotation narrative", elem_classes=["ai-panel"])
 
     with gr.Tab("📰 Watchlist News"):
         gr.Markdown("Daily rule: for each **holding**, only **today's** news is checked. "
@@ -268,14 +278,14 @@ with gr.Blocks(title="Chan Compass · US", **_style_kw) as demo:
             res_in = gr.Textbox(label="Ticker", placeholder="e.g. NVDA", scale=3)
             res_btn = gr.Button("🤖 Run research agent", variant="primary", scale=1)
         res_progress = gr.Markdown()
-        res_out = gr.Markdown()
+        res_out = gr.Markdown(elem_classes=["ai-panel"])
         gr.Markdown("**Report library** (auto + manual, stored on `/data`):",
                     elem_classes=["s2-footnote"])
         with gr.Row():
             rep_pick = gr.Dropdown(choices=research_agent.list_reports(),
                                    label="Saved reports", scale=3)
             rep_open = gr.Button("📂 Open report", scale=1)
-        rep_view = gr.Markdown()
+        rep_view = gr.Markdown(elem_classes=["ai-panel"])
 
     with gr.Tab("⏰ Automation"):
         gr.Markdown(paths.storage_status())

automation.py

@@ -91,13 +91,20 @@ def run_pipeline(tickers=None, force: bool = True) -> str:
                 known = set()
             current = set(df["Ticker"].tolist()) if df is not None and len(df) else set()
             new_tickers = sorted(current - known)
+            generated = set()
             for t in new_tickers[:5]:  # safety cap per run
                 _log(f"New ticker {t} → auto-generating research report…")
-                _, trace = research_agent.run_research(t, auto=True)
-                _log(f"Report for {t} done{' (+trace)' if trace else ''}.")
+                report, trace = research_agent.run_research(t, auto=True)
+                if report:
+                    generated.add(t)
+                    _log(f"Report for {t} done{' (+trace)' if trace else ''}.")
+                else:
+                    _log(f"Report for {t} postponed (sub-agents still loading) — "
+                         f"will retry on the next run.")
+            done_set = known | (current - (set(new_tickers) - generated))
             if current:
                 with open(known_path, "w", encoding="utf-8") as f:
-                    json.dump(sorted(known | current), f)
+                    json.dump(sorted(done_set), f)
         except Exception as e:
             _log(f"Auto-research skipped: {e}")
 

llm_local.py

@@ -40,15 +40,35 @@ DEFAULT_MODEL = "Qwen3-4B  · default — fast + smart, still ≤4B"
 _THINK_RE = re.compile(r"<think>.*?</think>", re.S)
 _NCPU = max(2, (os.cpu_count() or 4))
 
-WORKER_LABEL = {"fast": "Translator/Narrator sub-agent",
-                "deep": "Analyst sub-agent"}
+# One dedicated sub-agent per feature — independent locks, so Signals-Explain,
+# Rotation narrative, News briefs and Auto-Research never fight over a model.
+# Three tiny 1.7B instances share ONE GGUF file on disk (~2 GB RAM each) and
+# the 4B Analyst writes reports. Total ≈ 9 GB on a 32 GB Space.
+WORKER_LABEL = {
+    "translator": "Translator sub-agent (Signals · Explain)",
+    "narrator":   "Narrator sub-agent (Sector Rotation)",
+    "reporter":   "Reporter sub-agent (News · Research support)",
+    "analyst":    "Analyst sub-agent (Auto Research)",
+}
+
+
+def _mk(model):
+    return {"model": model, "llm": None, "lock": threading.Lock(),
+            "load_lock": threading.Lock(), "stage": "idle", "detail": "", "ts": None}
+
 
 WORKERS = {
-    "fast": {"model": FAST_MODEL, "llm": None, "lock": threading.Lock(),
-             "load_lock": threading.Lock(), "stage": "idle", "detail": "", "ts": None},
-    "deep": {"model": DEFAULT_MODEL, "llm": None, "lock": threading.Lock(),
-             "load_lock": threading.Lock(), "stage": "idle", "detail": "", "ts": None},
+    "translator": _mk(FAST_MODEL),
+    "narrator":   _mk(FAST_MODEL),
+    "reporter":   _mk(FAST_MODEL),
+    "analyst":    _mk(DEFAULT_MODEL),
 }
+# legacy aliases
+_ALIAS = {"fast": "translator", "deep": "analyst"}
+
+
+def _wk(worker: str) -> str:
+    return _ALIAS.get(worker, worker)
 
 _install_lock = threading.Lock()
 
@@ -128,7 +148,8 @@ def _ensure_llama_cpp(worker: str) -> str:
 
 
 # ─────────────────────── loading ───────────────────────
-def load_model(name: str, worker: str = "deep") -> str:
+def load_model(name: str, worker: str = "analyst") -> str:
+    worker = _wk(worker)
     """Load a GGUF into a worker slot. Non-blocking: if that worker is already
     installing/loading, returns its live stage instead of hanging the click."""
     w = WORKERS[worker]
@@ -157,11 +178,12 @@ def load_model(name: str, worker: str = "deep") -> str:
         try:
             _set_stage(worker, "loading model into RAM", name)
             w["llm"] = None
+            small = worker != "analyst"
             w["llm"] = Llama(
                 model_path=path,
-                n_ctx=4096 if worker == "fast" else 6144,
-                n_threads=_NCPU,
-                n_threads_batch=_NCPU,
+                n_ctx=4096 if small else 6144,
+                n_threads=(4 if small else _NCPU),   # leave headroom for parallel agents
+                n_threads_batch=(6 if small else _NCPU),
                 n_batch=512,
                 verbose=False,
             )
@@ -177,16 +199,16 @@ def load_model(name: str, worker: str = "deep") -> str:
 
 
 def auto_load_all():
-    """Startup: bring the fast sub-agent up first (small download, features go
-    live quickly), then the deep one. Runs in a background thread."""
-    load_model(WORKERS["fast"]["model"], worker="fast")
-    load_model(WORKERS["deep"]["model"], worker="deep")
+    """Startup: tiny agents first (one small GGUF download serves all three),
+    then the Analyst. Runs in a background thread."""
+    for key in ("translator", "narrator", "reporter", "analyst"):
+        load_model(WORKERS[key]["model"], worker=key)
 
 
 # ─────────────────────── status ───────────────────────
 def is_loaded(worker: str = None) -> bool:
     if worker:
-        return WORKERS[worker]["llm"] is not None
+        return WORKERS[_wk(worker)]["llm"] is not None
     return any(w["llm"] is not None for w in WORKERS.values())
 
 
@@ -196,7 +218,7 @@ def available() -> bool:
 
 def status() -> str:
     lines = []
-    for key in ("fast", "deep"):
+    for key in ("translator", "narrator", "reporter", "analyst"):
         w = WORKERS[key]
         label = WORKER_LABEL[key]
         if w["llm"] is not None:
@@ -222,9 +244,9 @@ def _messages(user: str, system: str):
 
 
 def chat(user: str, max_tokens: int = 500, temperature: float = 0.3,
-         system: str = DEFAULT_SYSTEM, worker: str = "fast") -> str:
+         system: str = DEFAULT_SYSTEM, worker: str = "translator") -> str:
     """Blocking chat on one sub-agent (used by pipeline/agent code)."""
-    w = WORKERS[worker]
+    w = WORKERS[_wk(worker)]; worker = _wk(worker)
     if w["llm"] is None:
         return ""
     if not w["lock"].acquire(timeout=180):
@@ -242,9 +264,9 @@ def chat(user: str, max_tokens: int = 500, temperature: float = 0.3,
 
 
 def chat_stream(user: str, max_tokens: int = 500, temperature: float = 0.3,
-                system: str = DEFAULT_SYSTEM, worker: str = "fast"):
+                system: str = DEFAULT_SYSTEM, worker: str = "translator"):
     """Streaming chat on one sub-agent — yields cumulative text immediately."""
-    w = WORKERS[worker]
+    w = WORKERS[_wk(worker)]; worker = _wk(worker)
     if w["llm"] is None:
         yield (f"⏳ {WORKER_LABEL[worker]} isn't ready yet — "
                f"stage: {w['stage']}. Check the **Model** tab.")
@@ -274,7 +296,7 @@ def quick_test() -> str:
     """Sanity check both sub-agents."""
     import time
     outs = []
-    for key in ("fast", "deep"):
+    for key in ("translator", "narrator", "reporter", "analyst"):
         if WORKERS[key]["llm"] is None:
             outs.append(f"{WORKER_LABEL[key]}: not loaded ({WORKERS[key]['stage']})")
             continue

news_watch.py

@@ -86,12 +86,13 @@ def _llm_brief(ticker: str, items: list) -> str:
         prompt = (
             f"You are an equity news analyst. Today's headlines for {ticker} "
             f"(a stock the user currently HOLDS):\n{heads}\n\n"
-            "In English, ≤90 words: 1) one-line summary; 2) tag the net read as "
-            "POSITIVE / NEGATIVE / NEUTRAL for the holding; 3) one suggested "
-            "action (e.g. 'no action', 'review stop level', 'watch earnings'). "
-            "Be specific, no disclaimers."
+            "In ENGLISH ONLY, write:\n"
+            "1) **Per-headline:** one short line per headline above — what it says "
+            "and why it matters (or 'noise') for the holding;\n"
+            "2) **Net read:** POSITIVE / NEGATIVE / NEUTRAL with one sentence why;\n"
+            "3) **Action:** one concrete suggestion. ≤180 words, no disclaimers."
         )
-        return llm_local.chat(prompt, max_tokens=240, worker="fast")
+        return llm_local.chat(prompt, max_tokens=420, worker="reporter")
     except Exception:
         return ""
 

research_agent.py

@@ -31,21 +31,27 @@ import pandas as pd
 
 import paths
 
-SECTIONS = [
-    ("Valuation", "Judge whether the multiples are cheap/fair/rich versus the "
-                  "growth and margins in evidence. Quote 2-3 numbers."),
-    ("Moat & supply-chain position", "Infer the company's competitive moat and "
-                  "where it sits in its industry value chain (who depends on it, "
-                  "who it depends on)."),
-    ("Bull case", "The strongest 3-point case FOR owning the stock, grounded in "
-                  "the evidence."),
-    ("Bear case", "The strongest 3-point case AGAINST owning it."),
-    ("Technical timing (Chan theory)", "Interpret the Chan engine verdict for a "
-                  "long-term holder: act now, wait, or exit — and the key price "
-                  "levels."),
-    ("Risks & verdict", "Top risks, then one-line verdict: Buy / Accumulate / "
-                  "Hold / Avoid, sized for a long-hold portfolio."),
+# Multi-agent split: the 4B Analyst writes the heavy sections while the 1.7B
+# Reporter writes market/technical sections IN PARALLEL on its own lock —
+# wall-clock time ≈ the slower of the two instead of their sum.
+ANALYST_SECTIONS = [
+    ("Valuation", "cheap/fair/rich vs the growth and margins in evidence; quote 2-3 numbers"),
+    ("Technology moat", "the company's technological moat and how defensible it is"),
+    ("Supply-chain map", "a markdown table with two columns 'Upstream suppliers' and "
+     "'Downstream customers/users': list 4-6 real companies on each side WITH stock "
+     "tickers in parentheses, e.g. TSMC (TSM); mark private companies (private)"),
+    ("Bull case", "strongest 3 points FOR owning it"),
+    ("Bear case", "strongest 3 points AGAINST owning it"),
 ]
+REPORTER_SECTIONS = [
+    ("Money flow & related tickers", "read the MONEY FLOW evidence: is capital "
+     "entering or leaving the stock and its sector; name the sector ETF and 2-4 "
+     "related tickers worth watching"),
+    ("Technical timing (Chan theory)", "interpret the CHAN ENGINE VERDICT for a "
+     "long-term holder: act now, wait, or exit, and the key price levels"),
+    ("Risks & verdict", "top risks, then one line: Buy / Accumulate / Hold / Avoid"),
+]
+SECTIONS = ANALYST_SECTIONS + REPORTER_SECTIONS  # kept for trace readability
 
 
 # ───────────────────────── trace plumbing ─────────────────────────
@@ -146,6 +152,38 @@ def t_chan(ticker: str) -> str:
         return ""
 
 
+def t_flows(ticker: str) -> str:
+    """Money-flow proxy (Δ% × dollar volume) for the ticker and its sector ETF,
+    1/5/20-day windows, plus related tickers via the sector mapping."""
+    try:
+        import data_us
+        import rotation as rot
+        out = []
+        d = data_us.load_level(ticker, "d")
+        for n, lab in ((1, "1D"), (5, "5D"), (20, "20D")):
+            st = rot._window_stats(d, n)
+            if st:
+                pct, dvol, flow = st
+                out.append(f"{ticker} {lab}: {pct:+.2%}, flow proxy "
+                           f"${flow/1e6:+,.0f}M on ${dvol/1e9:,.1f}B avg $vol")
+        sector = ""
+        try:
+            import yfinance as yf
+            sector = (yf.Ticker(ticker).info or {}).get("sector", "")
+        except Exception:
+            pass
+        etf = next((k for k, v in rot.SECTOR_ETFS.items() if v == sector), None)
+        if etf:
+            de = data_us.load_level(etf, "d")
+            st = rot._window_stats(de, 5)
+            if st:
+                out.append(f"Sector ETF {etf} ({sector}) 5D: {st[0]:+.2%}, "
+                           f"flow proxy ${st[2]/1e6:+,.0f}M")
+        return "\n".join(out)
+    except Exception:
+        return ""
+
+
 def t_news(ticker: str) -> str:
     try:
         import yfinance as yf
@@ -161,78 +199,108 @@ def t_news(ticker: str) -> str:
 
 
 # ───────────────────────── the agent ─────────────────────────
+def _gather_evidence(ticker: str, tr: "Trace", on_step=None) -> dict:
+    """Run all evidence tools IN PARALLEL (network-bound) — was serial before."""
+    from concurrent.futures import ThreadPoolExecutor
+    tools = {"fundamentals": t_fundamentals, "financials": t_financials,
+             "price": t_price, "chan_engine": t_chan, "flows": t_flows,
+             "news": t_news}
+    evidence = {}
+    with ThreadPoolExecutor(max_workers=6) as ex:
+        futs = {name: ex.submit(fn, ticker) for name, fn in tools.items()}
+        for name, fut in futs.items():
+            try:
+                evidence[name] = fut.result(timeout=40)
+            except Exception as e:
+                evidence[name] = ""
+                tr.log(f"TOOL:{name}", "tool_error", str(e))
+            brief = (evidence[name].get("plain", "")[:300]
+                     if isinstance(evidence[name], dict) else str(evidence[name])[:300])
+            tr.log(f"TOOL:{name}", "tool_result", brief or "(empty)")
+            if on_step:
+                on_step(name)
+    return evidence
+
+
+def _evidence_text(evidence: dict) -> str:
+    fund = evidence.get("fundamentals")
+    return (f"FUNDAMENTALS:\n{(fund.get('plain','') if isinstance(fund, dict) else '')[:1000]}\n\n"
+            f"QUARTERLY FINANCIALS:\n{str(evidence.get('financials'))[:380] or 'n/a'}\n\n"
+            f"PRICE ACTION:\n{evidence.get('price') or 'n/a'}\n\n"
+            f"MONEY FLOW:\n{str(evidence.get('flows'))[:420] or 'n/a'}\n\n"
+            f"CHAN ENGINE VERDICT:\n{str(evidence.get('chan_engine'))[:380] or 'n/a'}\n\n"
+            f"RECENT HEADLINES:\n{str(evidence.get('news'))[:380] or 'n/a'}")[:2500]
+
+
+def _sections_prompt(ticker: str, sections, ev_text: str) -> str:
+    sec_list = "\n".join(f"## {t} — {i}" for t, i in sections)
+    return (f"You are a buy-side equity analyst. Write part of a research note on "
+            f"{ticker} using ONLY the evidence below (write 'n/a' for missing facts, "
+            f"never invent numbers; company names in the supply-chain table may come "
+            f"from your own industry knowledge). Output EXACTLY these markdown "
+            f"sections, ≤70 words each, plain prose, ENGLISH ONLY, no disclaimers:\n"
+            f"{sec_list}\n\nEVIDENCE:\n{ev_text}")
+
+
+_STATIC_PLAN = ("1. Is the valuation justified by growth?\n"
+                "2. How durable is the technology moat and supply-chain position?\n"
+                "3. Where is capital flowing — into or out of the name and sector?\n"
+                "4. Is now a good technical entry for a long-term holder?")
+
+
 def run_research(ticker: str, auto: bool = False) -> tuple:
-    """Run the full agent. Returns (report_markdown, trace_path)."""
+    """Blocking multi-agent run (used by the daily pipeline).
+    Analyst (4B) and Reporter (1.7B) write their sections in PARALLEL.
+    Returns (report_markdown, trace_path). If no model is ready, returns
+    ('', '') so the caller can postpone instead of saving an empty report."""
+    import llm_local
     ticker = (ticker or "").strip().upper()
     if not ticker:
         return "Enter a ticker symbol first.", ""
+    if not (llm_local.is_loaded("analyst") or llm_local.is_loaded("reporter")):
+        return "", ""   # postpone — model still loading
     tr = Trace(ticker)
-    tr.log("PLAN", "llm_request", f"Draft research questions for {ticker}")
-    plan = _llm(f"You are a buy-side research agent. List the 4 most important "
-                f"questions to answer before a LONG-TERM investment in {ticker}. "
-                f"One line each, no preamble.", 200)
-    if not plan:
-        plan = ("1. Is the valuation justified by growth?\n2. How durable is the moat?\n"
-                "3. What breaks the bull thesis?\n4. Is now a good technical entry?")
-        tr.log("PLAN", "fallback", plan)
-    else:
-        tr.log("PLAN", "llm_response", plan)
-
-    evidence = {}
-    for name, fn in (("fundamentals", t_fundamentals), ("financials", t_financials),
-                     ("price", t_price), ("chan_engine", t_chan), ("news", t_news)):
-        tr.log(f"TOOL:{name}", "tool_call", f"{name}({ticker})")
-        try:
-            out = fn(ticker)
-        except Exception as e:
-            out = ""
-            tr.log(f"TOOL:{name}", "tool_error", f"{e}\n{traceback.format_exc(limit=1)}")
-        evidence[name] = out
-        brief = (out.get("plain", "")[:400] if isinstance(out, dict) else str(out)[:400])
-        tr.log(f"TOOL:{name}", "tool_result", brief or "(empty)")
-
-    fund = evidence["fundamentals"]
+    tr.log("PLAN", "static", _STATIC_PLAN)
+    evidence = _gather_evidence(ticker, tr)
+    fund = evidence.get("fundamentals")
     if isinstance(fund, dict) and not fund.get("ok"):
         tr.save()
         return f"⚠️ {fund.get('error', 'Could not fetch data.')}", ""
-
-    ev_text = (f"FUNDAMENTALS:\n{(fund.get('plain','') if isinstance(fund, dict) else '')[:1100]}\n\n"
-               f"QUARTERLY FINANCIALS:\n{str(evidence['financials'])[:450] or 'n/a'}\n\n"
-               f"PRICE ACTION:\n{evidence['price'] or 'n/a'}\n\n"
-               f"CHAN ENGINE VERDICT:\n{str(evidence['chan_engine'])[:420] or 'n/a'}\n\n"
-               f"RECENT HEADLINES:\n{str(evidence['news'])[:420] or 'n/a'}")[:2700]
-
-    # ONE LLM call for the whole analysis (six calls = six slow CPU prefills;
-    # one call with all sections in the prompt is ~5x faster on a Space).
-    parts = []
-    sec_list = "\n".join(f"## {t} — {i}" for t, i in SECTIONS)
-    tr.log("ANALYZE", "llm_request", f"single-pass report, {len(ev_text)} chars evidence")
-    txt = _llm(f"You are a buy-side equity analyst. Write a research note on {ticker} "
-               f"using ONLY the evidence below (write 'n/a' for missing facts, never "
-               f"invent numbers). Output EXACTLY these markdown sections, ≤70 words "
-               f"each, plain prose, no disclaimers:\n{sec_list}\n\nEVIDENCE:\n{ev_text}",
-               900)
-    tr.log("ANALYZE", "llm_response", txt[:600])
-    if txt:
-        parts.append(txt)
+    ev_text = _evidence_text(evidence)
+
+    import threading
+    parts = {}
+
+    def _write(slot, sections, worker, fallback_worker):
+        wk = worker if llm_local.is_loaded(worker) else fallback_worker
+        tr.log(f"ANALYZE:{slot}", "llm_request", f"worker={wk}")
+        txt = llm_local.chat(_sections_prompt(ticker, sections, ev_text),
+                             max_tokens=620 if slot == "analyst" else 360, worker=wk)
+        if txt.startswith("(") or txt.startswith("⏳"):
+            txt = ""
+        tr.log(f"ANALYZE:{slot}", "llm_response", txt[:500])
+        parts[slot] = txt
+
+    th = threading.Thread(target=_write,
+                          args=("reporter", REPORTER_SECTIONS, "reporter", "analyst"))
+    th.start()
+    _write("analyst", ANALYST_SECTIONS, "analyst", "reporter")
+    th.join(timeout=300)
 
     fund_md = fund.get("markdown", "") if isinstance(fund, dict) else ""
     stamp = dt.datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")
     head = (f"# {ticker} — Research Note{' (auto-generated)' if auto else ''}\n"
-            f"_{stamp} · Chan Compass research agent · local llama.cpp model_\n\n"
-            f"**Agent plan:**\n{plan}\n\n{fund_md}\n\n---\n")
-    if parts:
-        body = "\n\n".join(parts)
-    else:
-        body = ("## Analysis\n_Load a model in the **Model** tab to generate the written "
-                "sections — the agent gathered all evidence above and saved its trace._")
-    report = head + body + "\n\n---\n_Trace saved for the open-trace badge — see the Automation tab._"
-
-    tr.log("REPORT", "assembled", f"{len(report)} chars, {len(parts)} LLM sections")
+            f"_{stamp} · multi-agent: Analyst (Qwen3-4B) + Reporter (Qwen3-1.7B), "
+            f"both llama.cpp local_\n\n**Agent plan:**\n{_STATIC_PLAN}\n\n{fund_md}\n\n---\n")
+    body = "\n\n".join(p for p in (parts.get("analyst"), parts.get("reporter")) if p)
+    if not body:
+        tr.save()
+        return "", ""   # model produced nothing — postpone, never save a stub
+    report = head + body + "\n\n---\n_Agent trace saved — see the Automation tab._"
+    tr.log("REPORT", "assembled", f"{len(report)} chars")
     trace_path = tr.save()
     try:
-        rp = os.path.join(paths.REPORTS_DIR,
-                          f"{ticker}_{tr.t0.strftime('%Y%m%d')}.md")
+        rp = os.path.join(paths.REPORTS_DIR, f"{ticker}_{tr.t0.strftime('%Y%m%d')}.md")
         with open(rp, "w", encoding="utf-8") as f:
             f.write(report)
     except OSError:
@@ -274,81 +342,77 @@ def list_traces() -> str:
 
 # ───────────────────────── streaming UI runner ─────────────────────────
 def run_research_stream(ticker: str):
-    """Generator for the Gradio UI: yields live progress, then streams the
-    LLM analysis token-by-token, then saves report + trace (same artifacts
-    as run_research)."""
+    """Generator for the UI. Multi-agent: evidence tools run in parallel; the
+    1.7B Reporter writes its sections in a background thread while the 4B
+    Analyst STREAMS its sections live; never saves a 'model busy' stub."""
+    import threading
+    import llm_local
     ticker = (ticker or "").strip().upper()
     if not ticker:
         yield "Enter a ticker symbol first.", ""
         return
     tr = Trace(ticker)
-    log_lines = [f"### 🤖 Research agent · {ticker}"]
+    log_lines = [f"### 🤖 Multi-agent research · {ticker}"]
 
     def show(msg):
         log_lines.append(f"- {msg}")
         return "\n".join(log_lines)
 
-    yield show("**PLAN** — drafting research questions…"), ""
-    plan = _llm(f"List the 4 most important questions to answer before a LONG-TERM "
-                f"investment in {ticker}. One line each, no preamble.", 140)
-    if not plan:
-        plan = ("1. Is the valuation justified by growth?\n2. How durable is the moat?\n"
-                "3. What breaks the bull thesis?\n4. Is now a good technical entry?")
-    tr.log("PLAN", "llm_response", plan)
-    yield show("Plan ready ✓"), ""
+    tr.log("PLAN", "static", _STATIC_PLAN)
+    yield show("**PLAN** ready ✓ · **TOOLS** — gathering 6 evidence sources in parallel…"), ""
+    evidence = _gather_evidence(ticker, tr)
+    yield show("Evidence in: fundamentals · financials · price · money flow · Chan engine · news ✓"), ""
 
-    evidence = {}
-    for name, fn, label in (("fundamentals", t_fundamentals, "fundamentals & valuation"),
-                            ("financials", t_financials, "quarterly financials"),
-                            ("price", t_price, "price action stats"),
-                            ("chan_engine", t_chan, "Chan engine verdict"),
-                            ("news", t_news, "recent headlines")):
-        yield show(f"**TOOL** — gathering {label}…"), ""
-        try:
-            evidence[name] = fn(ticker)
-        except Exception as e:
-            evidence[name] = ""
-            tr.log(f"TOOL:{name}", "tool_error", str(e))
-        brief = (evidence[name].get("plain", "")[:300] if isinstance(evidence[name], dict)
-                 else str(evidence[name])[:300])
-        tr.log(f"TOOL:{name}", "tool_result", brief or "(empty)")
-
-    fund = evidence["fundamentals"]
+    fund = evidence.get("fundamentals")
     if isinstance(fund, dict) and not fund.get("ok"):
         tr.save()
         yield show(f"⚠️ {fund.get('error', 'Data fetch failed.')}"), ""
         return
-
-    ev_text = (f"FUNDAMENTALS:\n{(fund.get('plain','') if isinstance(fund, dict) else '')[:1100]}\n\n"
-               f"QUARTERLY FINANCIALS:\n{str(evidence['financials'])[:450] or 'n/a'}\n\n"
-               f"PRICE ACTION:\n{evidence['price'] or 'n/a'}\n\n"
-               f"CHAN ENGINE VERDICT:\n{str(evidence['chan_engine'])[:420] or 'n/a'}\n\n"
-               f"RECENT HEADLINES:\n{str(evidence['news'])[:420] or 'n/a'}")[:2700]
-    yield show("**ANALYZE** — writing the report (streaming)…"), ""
-
+    ev_text = _evidence_text(evidence)
     fund_md = fund.get("markdown", "") if isinstance(fund, dict) else ""
     stamp = dt.datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")
-    head = (f"# {ticker} — Research Note\n_{stamp} · Chan Compass research agent · "
-            f"local llama.cpp model_\n\n**Agent plan:**\n{plan}\n\n{fund_md}\n\n---\n")
-    sec_list = "\n".join(f"## {t} — {i}" for t, i in SECTIONS)
-    prompt = (f"You are a buy-side equity analyst. Write a research note on {ticker} "
-              f"using ONLY the evidence below (write 'n/a' for missing facts, never "
-              f"invent numbers). Output EXACTLY these markdown sections, ≤70 words "
-              f"each, plain prose, no disclaimers:\n{sec_list}\n\nEVIDENCE:\n{ev_text}")
-    tr.log("ANALYZE", "llm_request", f"single-pass report, {len(ev_text)} chars evidence")
-    body = ""
-    try:
-        import llm_local
-        wk = "deep" if llm_local.is_loaded("deep") else "fast"
-        for acc in llm_local.chat_stream(prompt, max_tokens=900, worker=wk):
-            body = acc
-            yield "\n".join(log_lines), head + body
-    except Exception as e:
-        body = f"(model error: {e})"
-    if not body.strip() or body.startswith("⏳"):
-        body = ("## Analysis\n_Model not available — evidence gathered above; "
-                "trace saved._\n\n" + body)
-    tr.log("ANALYZE", "llm_response", body[:600])
+    head = (f"# {ticker} — Research Note\n_{stamp} · multi-agent: Analyst (Qwen3-4B) "
+            f"+ Reporter (Qwen3-1.7B), both llama.cpp local_\n\n"
+            f"**Agent plan:**\n{_STATIC_PLAN}\n\n{fund_md}\n\n---\n")
+
+    # Reporter sub-agent works in parallel on its own lock
+    side = {"txt": ""}
+
+    def _side():
+        wk = "reporter" if llm_local.is_loaded("reporter") else "analyst"
+        t = llm_local.chat(_sections_prompt(ticker, REPORTER_SECTIONS, ev_text),
+                           max_tokens=360, worker=wk)
+        side["txt"] = "" if (t.startswith("(") or t.startswith("⏳")) else t
+        tr.log("ANALYZE:reporter", "llm_response", side["txt"][:400])
+
+    th = None
+    if llm_local.is_loaded("reporter") or llm_local.is_loaded("analyst"):
+        th = threading.Thread(target=_side, daemon=True)
+        th.start()
+        yield show("**Reporter sub-agent** writing money-flow / Chan timing / verdict "
+                   "in parallel…"), head
+    # Analyst streams the main sections live
+    main = ""
+    wk_main = "analyst" if llm_local.is_loaded("analyst") else "reporter"
+    if llm_local.is_loaded(wk_main):
+        yield show(f"**Analyst sub-agent** streaming valuation / moat / supply-chain "
+                   f"map / bull-bear…"), head
+        for acc in llm_local.chat_stream(
+                _sections_prompt(ticker, ANALYST_SECTIONS, ev_text),
+                max_tokens=620, worker=wk_main):
+            if acc.startswith("⏳") or acc.startswith("("):
+                continue
+            main = acc
+            yield "\n".join(log_lines), head + main
+    tr.log("ANALYZE:analyst", "llm_response", main[:500])
+    if th is not None:
+        th.join(timeout=240)
+    body = "\n\n".join(p for p in (main, side["txt"]) if p)
+    if not body:
+        tr.save()
+        yield show("⚠️ Sub-agents not ready yet (still loading) — evidence gathered "
+                   "above; try again in a minute. Nothing was saved."), head
+        return
     report = head + body + "\n\n---\n_Agent trace saved — see the Automation tab._"
     trace_path = tr.save()
     try:

rotation.py

@@ -150,6 +150,6 @@ def llm_narrative(df_1d, df_5d, df_20d) -> str:
             "trend (rotation vs one-day noise); 3) one actionable watch item. "
             "No disclaimers.\n\nDATA:\n" + brief
         )
-        return llm_local.chat(prompt, max_tokens=380, worker="fast")
+        return llm_local.chat(prompt, max_tokens=380, worker="narrator")
     except Exception as e:
         return f"**Raw read:**\n{brief}\n\n_(LLM unavailable: {e})_"