cyb50-quant/research/dragon/v2/dragon_validate.py

from __future__ import annotations

from pathlib import Path

import pandas as pd

from dragon_indicators import DragonIndicatorConfig, DragonIndicatorEngine
from dragon_workbook import DragonWorkbook


KDJ_BUY_MARKER = chr(0x91D1)
KDJ_SELL_MARKER = chr(0x6B7B)
QL_BUY_MARKER = "B"
QL_SELL_MARKER = "S"


def _find_workbook(base_dir: Path) -> Path:
    matches = sorted(base_dir.glob("*.xlsx"))
    if not matches:
        raise FileNotFoundError(f"No workbook found in {base_dir}")
    return matches[0]


def _marker_match(workbook_value: object, computed_value: bool, positive: str) -> bool:
    if workbook_value is None:
        return not computed_value
    if not isinstance(workbook_value, str):
        return not computed_value
    return (positive in workbook_value) == bool(computed_value)


def _explicit_marker(value: object, allowed: set[str]) -> str | None:
    if not isinstance(value, str):
        return None
    stripped = value.strip()
    return stripped if stripped in allowed else None


def main() -> None:
    base_dir = Path(__file__).resolve().parent
    workbook_path = _find_workbook(base_dir)
    workbook = DragonWorkbook(workbook_path)
    split_events = workbook.split_layers()
    annual_summary = workbook.load_annual_summary()

    config = DragonIndicatorConfig(start_date="2015-01-01", end_date="2026-01-31")
    engine = DragonIndicatorEngine(config)
    raw_df = engine.fetch_daily_data()
    indicators = engine.compute(raw_df)

    indicators_out = indicators.reset_index().rename(columns={"index": "date"})
    indicators_out["date"] = pd.to_datetime(indicators_out["date"]).dt.date.astype(str)
    indicators_out.to_csv(base_dir / "dragon_indicator_snapshot.csv", index=False, encoding="utf-8-sig")

    benchmark_df = pd.DataFrame(
        [
            {
                "date": event.date.isoformat(),
                "side": event.side,
                "layer": event.layer,
                "signal_reason": event.signal_reason,
                "index_close": event.index_close,
                "prev_index": event.prev_index,
                "capital": event.capital,
                "pnl": event.pnl,
                "kdj": event.kdj,
                "ql": event.ql,
                "note": event.note,
            }
            for event in split_events
        ]
    )
    benchmark_df.to_csv(base_dir / "dragon_workbook_layers.csv", index=False, encoding="utf-8-sig")

    workbook_markers = benchmark_df[["date", "kdj", "ql"]].drop_duplicates(subset=["date"]).copy()
    workbook_markers["kdj_marker"] = workbook_markers["kdj"].apply(
        lambda value: _explicit_marker(value, {KDJ_BUY_MARKER, KDJ_SELL_MARKER})
    )
    workbook_markers["ql_marker"] = workbook_markers["ql"].apply(
        lambda value: _explicit_marker(value, {QL_BUY_MARKER, QL_SELL_MARKER})
    )

    compare = workbook_markers.merge(
        indicators_out[
            ["date", "a1", "b1", "c1", "kdj_buy", "kdj_sell", "ql_buy", "ql_sell", "close"]
        ],
        on="date",
        how="left",
    )
    compare["kdj_buy_match"] = compare.apply(
        lambda row: _marker_match(row["kdj_marker"], bool(row["kdj_buy"]), KDJ_BUY_MARKER),
        axis=1,
    )
    compare["kdj_sell_match"] = compare.apply(
        lambda row: _marker_match(row["kdj_marker"], bool(row["kdj_sell"]), KDJ_SELL_MARKER),
        axis=1,
    )
    compare["ql_buy_match"] = compare.apply(
        lambda row: _marker_match(row["ql_marker"], bool(row["ql_buy"]), QL_BUY_MARKER),
        axis=1,
    )
    compare["ql_sell_match"] = compare.apply(
        lambda row: _marker_match(row["ql_marker"], bool(row["ql_sell"]), QL_SELL_MARKER),
        axis=1,
    )
    compare.to_csv(base_dir / "dragon_signal_alignment.csv", index=False, encoding="utf-8-sig")

    real_events = benchmark_df[benchmark_df["layer"] == "real_trade"]
    aux_events = benchmark_df[benchmark_df["layer"] == "aux_signal"]
    kdj_rows = compare[compare["kdj_marker"].notna()]
    ql_rows = compare[compare["ql_marker"].notna()]
    kdj_sell_mismatch_dates = compare.loc[
        compare["kdj_marker"].eq(KDJ_SELL_MARKER) & ~compare["kdj_sell_match"], "date"
    ].tolist()
    ql_buy_mismatch_dates = compare.loc[
        compare["ql_marker"].eq(QL_BUY_MARKER) & ~compare["ql_buy_match"], "date"
    ].tolist()

    lines = [
        "# Dragon V2 Validation",
        "",
        f"- Source workbook: `{workbook_path.name}`",
        f"- Indicator snapshot rows: `{len(indicators_out)}`",
        f"- Workbook layered events: `{len(benchmark_df)}`",
        f"- Real trade events: `{len(real_events)}`",
        f"- Auxiliary events: `{len(aux_events)}`",
        f"- Annual summary rows: `{len(annual_summary)}`",
        "",
        "## Formula Scope",
        "- `A1`: EMA(8) vs EMA(EMA(8),20) normalized spread",
        "- `B1`: ((Y2 - Y3) / 100), where Y2/Y3 come from 38-day RSV smoothed by SMA(5,1) and SMA(10,1)",
        "- `C1`: (Y2 + Y3) / 2",
        "- `QL phoenix line`: approximated from the existing Dragon code as `B = CROSS(close, upper_line)` and `S = CROSS(lower_line, close)`",
        "",
        "## Alignment",
        f"- KDJ rows in workbook: `{len(kdj_rows)}`",
        f"- QL rows in workbook: `{len(ql_rows)}`",
        f"- KDJ buy marker matches: `{int(kdj_rows['kdj_buy_match'].sum())}/{len(kdj_rows)}`",
        f"- KDJ sell marker matches: `{int(kdj_rows['kdj_sell_match'].sum())}/{len(kdj_rows)}`",
        f"- QL buy marker matches: `{int(ql_rows['ql_buy_match'].sum())}/{len(ql_rows)}`",
        f"- QL sell marker matches: `{int(ql_rows['ql_sell_match'].sum())}/{len(ql_rows)}`",
        f"- KDJ sell mismatch dates: `{kdj_sell_mismatch_dates}`",
        f"- QL buy mismatch dates: `{ql_buy_mismatch_dates}`",
        "",
        "## Output Files",
        "- `dragon_indicator_snapshot.csv`",
        "- `dragon_workbook_layers.csv`",
        "- `dragon_signal_alignment.csv`",
        "",
        "## Notes",
        "- This validation stage checks indicator reconstruction and marker alignment only.",
        "- It does not yet implement the full Dragon entry/exit rule tree from the workbook narrative.",
        "- Repeated SELL rows after exit and repeated BUY rows during holding remain auxiliary signals by confirmed user semantics.",
    ]
    (base_dir / "dragon_validation.md").write_text("\n".join(lines) + "\n", encoding="utf-8")


if __name__ == "__main__":
    main()