signal_runner.py

"""
signal_runner.py — run the (unchanged) Chan multi-level engine over a US ticker pool.

Levels per ticker:  monthly / weekly  (resampled from 10y daily)
                    daily             (yfinance 1d)
                    60m / 30m / 15m / 5m  (yfinance intraday)
                    1m                (not available beyond 7 days on Yahoo → skipped;
                                       MultiLevelChan degrades gracefully)
"""
from __future__ import annotations

import os
import traceback

import pandas as pd

import chan_glue  # noqa: F401  (wires engine into chan_multilevel, installs cached factory)
# Import names straight from their home module — never rely on re-exports,
# so a stale chan_glue.py on the Space can't break startup.
from chan_multilevel import MultiLevelChan, resample_weekly, resample_monthly
import chan_enhance
import data_us

DEFAULT_POOL = ["AAPL", "MSFT", "NVDA", "TSLA", "AMZN", "GOOGL", "META", "AMD", "NFLX", "JPM"]

# ── LONG-HOLD mode (user requirement for the US version) ────────────────
# Operating level = weekly: ride the pivot uplift, don't get shaken out early.
#   mode='long'                     → daily S1/S2 in a big uptrend → HOLD (armed)
#   require_sublevel_sell_confirm   → unconfirmed daily sells in an uptrend → HOLD
# Real exits that still fire: S3 (pivot breakdown), structural stop,
# and the armed exit line once the nested-interval top is confirmed.
MultiLevelChan.CFG["mode"] = "long"
MultiLevelChan.CFG["require_sublevel_sell_confirm"] = True

STOP_MAX_LOSS = 0.05  # same global loss cap as the user's backtest


def _structural_stop(kind: str, res) -> float | None:
    """Simplified invalidation price, lifted from the user's backtest logic:
    B1 → divergence low; B2 → retest low / B1 anchor; B3 → daily pivot ZD.
    Capped so a single position can never lose much more than STOP_MAX_LOSS."""
    sig = res.daily.signal if (res and res.daily) else None
    ex = (sig.extras if sig is not None else None) or {}
    close_p = float(res.cur_price)
    stop = None
    if kind == "B1":
        stop = ex.get("c_new_low") or ex.get("b1_price")
    elif kind == "B2":
        stop = ex.get("cur_low") or ex.get("b1_price")
    elif kind == "B3":
        stop = res.daily.zd if (res.daily and res.daily.zd) else ex.get("pull_low")
    if stop is None:
        stop = close_p * (1 - STOP_MAX_LOSS)
    stop = max(min(float(stop), close_p * 0.999), close_p * (1 - STOP_MAX_LOSS))
    return round(stop, 2)


def _next_day_plan(res) -> dict:
    """The simplified answer: do I buy/sell tomorrow, the exact BUY POINT, the
    acceptable entry zone, and the invalidation price. All prices come straight
    from the engine's signal.extras (same source as the user's backtest)."""
    kind, act = res.final_kind, res.action
    px = float(res.cur_price)
    if act == "BUY" and kind in ("B1", "B2", "B3"):
        stop = _structural_stop(kind, res)
        # The BUY POINT = the engine's structural level for this signal type:
        #   B1 divergence low · B2 retest low · B3 pivot upper band (ZG).
        sig = res.daily.signal if res.daily else None
        ex = (sig.extras if sig else None) or {}
        if kind == "B3":
            point = float(res.daily.zg) if (res.daily and res.daily.zg) else stop
        elif kind == "B1":
            point = float(ex.get("c_new_low") or stop)
        else:  # B2
            point = float(ex.get("cur_low") or ex.get("b1_price") or stop)
        lo = min(point, stop) if kind == "B3" else point
        hi = px * 1.015
        return {"plan": "🟢 BUY tomorrow at open",
                "point": f"${point:,.2f}",
                "zone": f"${min(lo, hi):,.2f} – ${hi:,.2f}",
                "stop": f"${stop:,.2f}",
                "hint": f"Long-hold entry near ${point:,.2f}; keep until S3 / stop / armed exit."}
    if act == "SELL":
        hint = {"STOP": "Structural stop hit — exit to protect capital.",
                "S3": "Pivot breakdown (S3) — the long-hold exit signal. Exit, don't average down."}
        return {"plan": "🔴 SELL tomorrow at open", "point": f"≈ ${px:,.2f}",
                "zone": f"≈ ${px:,.2f}", "stop": "—",
                "hint": hint.get(kind, "Confirmed top (divergence verified at sub-levels) — take profit.")}
    if act == "HOLD":
        if res.sell_armed and res.arm_zd:
            return {"plan": "🟡 HOLD (exit line armed)", "point": "—", "zone": "—",
                    "stop": f"${float(res.arm_zd):,.2f}",
                    "hint": f"Keep holding; sell only if price closes below ${float(res.arm_zd):,.2f}."}
        return {"plan": "🟡 HOLD", "point": "—", "zone": "—", "stop": "—",
                "hint": "Trend intact — long-hold, ignore daily noise."}
    # WAIT
    if kind:
        hint = (f"{kind} signal on the daily chart but NOT yet confirmed down the "
                f"nested sub-levels (60m→30m→15m→5m) — wait, don't chase.")
    elif res.blocked_reason:
        hint = "Signal blocked by a higher-timeframe gate (weekly/monthly direction). Stay out."
    else:
        wk = TREND_EN.get(res.weekly.trend if res.weekly else "", "?")
        dy = TREND_EN.get(res.daily.trend if res.daily else "", "?")
        hint = f"No buy/sell point today (weekly {wk}, daily {dy}). Stay in cash / keep watching."
    return {"plan": "⚪ WAIT", "point": "—", "zone": "—", "stop": "—", "hint": hint}

import paths

OUT_DIR = paths.OUTPUT_DIR

ACTION_BADGE = {"BUY": "🟢 BUY", "SELL": "🔴 SELL", "HOLD": "🟡 HOLD", "WATCH": "⚪ WATCH"}
KIND_EN = {
    "B1": "B1 · 1st buy (trend-end divergence)",
    "B2": "B2 · 2nd buy (higher-low retest)",
    "B3": "B3 · 3rd buy (pivot breakout retest)",
    "S1": "S1 · 1st sell (top divergence)",
    "S2": "S2 · 2nd sell (lower-high rebound)",
    "S3": "S3 · 3rd sell (pivot breakdown)",
    "STOP": "STOP · structural stop-loss",
    "": "—",
}
TREND_EN = {"up_trend": "Up", "down_trend": "Down", "consolidation": "Range",
            "expanding": "Expanding", "unknown": "?", "": "?"}


def stock_raw_read(ticker: str) -> str:
    """Plain-English factual snapshot of one ticker's Chan verdict — the
    deterministic 'Raw read' the Signals AI narrative summarizes (mirrors the
    Sector-Rotation pattern: raw read → agent → narrative)."""
    row = automation_state_row(ticker)
    if not row:
        return ""
    parts = [
        f"{ticker}: close {row['Close']}, signal {row['Signal']}, "
        f"confidence {row['Confidence']}.",
        f"Plan tomorrow: {row['Tomorrow']}.",
    ]
    if row.get("Buy point") not in ("—", None):
        parts.append(f"Buy point {row['Buy point']}, zone {row['Buy zone']}, "
                     f"invalid below {row['Invalid below']}.")
    parts.append(f"Note: {row['Note']}")
    return " ".join(parts)


def automation_state_row(ticker: str):
    import automation
    df = automation.STATE.get("signals_df")
    if df is None or "Ticker" not in getattr(df, "columns", []):
        return None
    m = df[df["Ticker"] == ticker]
    return m.iloc[0].to_dict() if len(m) else None


def analyze_one(ticker: str, force: bool = False):
    """Run the simplified long-hold Chan analysis for one ticker.
    Full nested-interval set: monthly/weekly (resampled) + daily +
    60m/30m/15m/5m/1m confirmation (区间套). Deeper sub-levels = more precise
    buy/sell points; any missing level (e.g. 1m beyond 7 days) is skipped."""
    dfs = data_us.load_levels(ticker, data_us.FULL_LEVELS, force=force)
    d = dfs["d"]
    if d is None or len(d) < 60:
        return None, f"{ticker}: not enough daily history ({0 if d is None else len(d)} bars)."
    w = resample_weekly(d)
    m = resample_monthly(d)

    ml = MultiLevelChan(
        df_daily=d, df_weekly=w, df_monthly=m,
        df_60m=dfs.get("60m"), df_30m=dfs.get("30m"),
        df_15m=dfs.get("15m"), df_5m=dfs.get("5m"),
        df_1m=dfs.get("1m"),   # finest nested-interval level when Yahoo has it
        code=ticker, strict=True,
    )
    res = ml.analyze()
    if res is None:
        return None, f"{ticker}: analysis returned no result (insufficient structure)."

    enh = chan_enhance.predict_enhance(res)
    weight = enh.get("suggest_weight")
    plan = _next_day_plan(res)
    row = {
        "Ticker": ticker,
        "Tomorrow": plan["plan"],
        "Buy point": plan["point"],
        "Buy zone": plan["zone"],
        "Invalid below": plan["stop"],
        "Signal": KIND_EN.get(res.final_kind, res.final_kind or "—"),
        "Confidence": res.confidence,
        "Close": f"${res.cur_price:,.2f}",
        "Weight": (f"{weight:.2f}" if weight else "—"),
        "Note": plan["hint"],
        "_action_raw": res.action,
        "_kind_raw": res.final_kind,
        "_date": res.analysis_date.strftime("%Y-%m-%d"),
    }
    detail = res.explain()
    extra_lines = []
    for k in ("l16_note", "l37_note", "evo_hint", "l92_warn"):
        if enh.get(k):
            extra_lines.append("  " + enh[k])
    if extra_lines:
        detail += "\n  ── 增强提示 (chan_enhance) ──\n" + "\n".join(extra_lines)
    return row, detail


def run_signals(tickers=None, force: bool = False):
    """Run the whole pool. Returns (DataFrame, {ticker: detail}, summary_str)."""
    tickers = [t.strip().upper() for t in (tickers or DEFAULT_POOL) if t.strip()]
    try:  # parallel download phase (IO-bound) — analysis stays CPU-only, no LLM
        data_us.prefetch(tickers, data_us.FULL_LEVELS, force=force, budget_s=60)
    except Exception:
        pass
    rows, details, errors = [], {}, []
    for t in tickers:
        try:
            row, detail = analyze_one(t, force=force)
            if row is None:
                errors.append(detail)
                continue
            rows.append(row)
            details[t] = detail
        except Exception as e:
            traceback.print_exc()
            errors.append(f"{t}: {e}")
    if rows:
        df = pd.DataFrame(rows)
        order = {"BUY": 0, "SELL": 1, "HOLD": 2, "WATCH": 3}
        df["_o"] = df["_action_raw"].map(order).fillna(9)
        df = df.sort_values(["_o", "Ticker"]).drop(columns=["_o"]).reset_index(drop=True)
        show = df.drop(columns=[c for c in df.columns if c.startswith("_")])
        try:
            df.to_csv(os.path.join(OUT_DIR, "signals_latest.csv"), index=False)
        except Exception:
            pass
    else:
        show = pd.DataFrame(columns=["Ticker", "Tomorrow", "Buy point", "Buy zone", "Invalid below", "Signal", "Confidence", "Close"])
    n_buy = sum(1 for r in rows if r["_action_raw"] == "BUY")
    n_sell = sum(1 for r in rows if r["_action_raw"] == "SELL")
    asof = rows[0]["_date"] if rows else "—"
    summary = (f"Analyzed {len(rows)}/{len(tickers)} tickers · as of {asof} · "
               f"{n_buy} BUY · {n_sell} SELL")
    if errors:
        summary += f" · {len(errors)} skipped"
    return show, details, summary, errors