scripts/backtest_c19_pattern_mining.py

#!/usr/bin/env python3
"""C19: dry-run candlestick + SAX pattern-mining features.

This is intentionally experimental. It does not modify production
FEATURE_COLUMNS. Promotion requires every tested stock to improve by at least
3 percentage points in no-lookahead walk-forward validation.
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

try:
    from dotenv import load_dotenv

    load_dotenv(ROOT / ".env")
except ImportError:
    pass

import warnings

warnings.filterwarnings("ignore")

import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier

from models.predictor import FEATURE_COLUMNS, _build_features
from scripts.improvement_harness import (
    DEFAULT_STOCKS,
    EXTENDED_STOCKS,
    LABEL_HORIZON,
    MIN_TRAIN,
    RF_PARAMS,
    STEP,
    build_triple_barrier_labels,
    compute_metrics,
)


PASS_PER_STOCK_ACCURACY_DELTA = 3.0
MIN_PATTERN_OBS = 5

RAW_PATTERN_FEATURES = [
    "candle_body_ratio",
    "upper_shadow_ratio",
    "lower_shadow_ratio",
    "close_position",
    "gap_pct",
    "range_pct",
    "body_pct",
    "range_expansion_5d",
    "volume_price_pressure",
    "inside_bar",
    "outside_bar",
    "bullish_engulfing",
    "bearish_engulfing",
    "candle_dir",
    "candle_pattern_3",
    "candle_pattern_5",
    "sax_pattern_5",
    "sax_pattern_10",
    "sax_momentum_5",
    "sax_reversal_5",
]

PATTERN_ID_COLUMNS = [
    "candle_pattern_3",
    "candle_pattern_5",
    "sax_pattern_5",
    "sax_pattern_10",
]


def _safe_div(num: pd.Series, den: pd.Series) -> pd.Series:
    return (num / den.replace(0, np.nan)).replace([np.inf, -np.inf], np.nan)


def _encode_base(values: pd.Series, base: int, window: int) -> pd.Series:
    encoded = pd.Series(0.0, index=values.index)
    for offset in range(window):
        encoded += values.shift(offset).fillna(0).astype(int) * (base ** offset)
    return encoded


def build_pattern_features(df: pd.DataFrame) -> pd.DataFrame:
    """Build leakage-safe raw pattern features from current/past OHLCV only."""
    out = pd.DataFrame(index=df.index)
    open_ = df["open"].astype(float)
    high = df["high"].astype(float)
    low = df["low"].astype(float)
    close = df["close"].astype(float)
    volume = df.get("volume", pd.Series(0.0, index=df.index)).astype(float)

    span = (high - low).abs().replace(0, np.nan)
    body = close - open_
    body_abs = body.abs()
    upper_shadow = high - np.maximum(open_, close)
    lower_shadow = np.minimum(open_, close) - low

    out["candle_body_ratio"] = _safe_div(body, span).clip(-1.0, 1.0)
    out["upper_shadow_ratio"] = _safe_div(upper_shadow, span).clip(0.0, 1.0)
    out["lower_shadow_ratio"] = _safe_div(lower_shadow, span).clip(0.0, 1.0)
    out["close_position"] = _safe_div(close - low, span).clip(0.0, 1.0)
    out["gap_pct"] = close.pct_change().fillna(0.0).clip(-0.2, 0.2)
    out["range_pct"] = _safe_div(high - low, close).clip(0.0, 0.25)
    out["body_pct"] = _safe_div(body_abs, close).clip(0.0, 0.25)

    range_mean = out["range_pct"].rolling(5, min_periods=2).mean().shift(1)
    out["range_expansion_5d"] = _safe_div(out["range_pct"], range_mean).fillna(1.0).clip(0.0, 5.0)

    volume_z = (volume - volume.rolling(20, min_periods=5).mean()) / volume.rolling(20, min_periods=5).std().replace(0, np.nan)
    out["volume_price_pressure"] = (out["candle_body_ratio"].fillna(0.0) * volume_z.fillna(0.0)).clip(-5.0, 5.0)

    prev_high = high.shift(1)
    prev_low = low.shift(1)
    prev_open = open_.shift(1)
    prev_close = close.shift(1)
    prev_body = prev_close - prev_open

    out["inside_bar"] = ((high <= prev_high) & (low >= prev_low)).astype(float)
    out["outside_bar"] = ((high >= prev_high) & (low <= prev_low)).astype(float)
    out["bullish_engulfing"] = ((body > 0) & (prev_body < 0) & (close >= prev_open) & (open_ <= prev_close)).astype(float)
    out["bearish_engulfing"] = ((body < 0) & (prev_body > 0) & (open_ >= prev_close) & (close <= prev_open)).astype(float)

    candle_dir = np.where(body > span * 0.1, 2, np.where(body < -span * 0.1, 0, 1))
    out["candle_dir"] = pd.Series(candle_dir, index=df.index).fillna(1).astype(float)
    out["candle_pattern_3"] = _encode_base(out["candle_dir"], base=3, window=3)
    out["candle_pattern_5"] = _encode_base(out["candle_dir"], base=3, window=5)

    ret_1d = close.pct_change().fillna(0.0)
    rolling_vol = ret_1d.rolling(20, min_periods=5).std().shift(1).fillna(0.01).clip(lower=0.002)
    sax_symbol = np.where(ret_1d > rolling_vol * 0.35, 2, np.where(ret_1d < -rolling_vol * 0.35, 0, 1))
    sax = pd.Series(sax_symbol, index=df.index).astype(float)
    out["sax_pattern_5"] = _encode_base(sax, base=3, window=5)
    out["sax_pattern_10"] = _encode_base(sax, base=3, window=10)
    out["sax_momentum_5"] = (sax.rolling(5, min_periods=1).sum() - 5).clip(-5, 5)
    out["sax_reversal_5"] = (sax - sax.shift(4)).fillna(0.0).clip(-2, 2)

    return out[RAW_PATTERN_FEATURES].replace([np.inf, -np.inf], np.nan).fillna(0.0)


def fetch_validation_df(stock_no: str) -> pd.DataFrame | None:
    """Load only features needed by current production columns plus OHLCV patterns."""
    from data.fetcher import fetch_cross_asset_tw
    from indicators.technical import add_all_indicators, add_cross_asset_tw
    from services.predictor_service import _fetch_with_cache

    df = _fetch_with_cache(stock_no, months=24)
    if df is None or df.empty:
        return None

    df = add_all_indicators(df)
    start, end = str(df["date"].min()), str(df["date"].max())
    result = fetch_cross_asset_tw(start, end)
    taiex, usdtwd = result[0], result[1]
    sox = result[2] if len(result) > 2 else None
    tnx = result[3] if len(result) > 3 else None
    return add_cross_asset_tw(df, taiex, usdtwd, sox_close=sox, tnx_close=tnx)


def _pattern_stats_for_fold(
    pattern_df: pd.DataFrame,
    labels: np.ndarray,
    train_idx: np.ndarray,
    test_idx: np.ndarray,
) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Create fold-local pattern stats without using test labels."""
    train_stats = pd.DataFrame(index=train_idx)
    test_stats = pd.DataFrame(index=test_idx)

    y_train = labels[train_idx]
    valid_train_idx = train_idx[~np.isnan(y_train)]
    valid_y = labels[valid_train_idx].astype(int)
    default_up = float(np.mean(valid_y == 1)) if len(valid_y) else 0.0
    default_down = float(np.mean(valid_y == -1)) if len(valid_y) else 0.0
    default_mean = float(np.mean(valid_y)) if len(valid_y) else 0.0

    for col in PATTERN_ID_COLUMNS:
        counts: dict[int, int] = {}
        up_counts: dict[int, int] = {}
        down_counts: dict[int, int] = {}
        sums: dict[int, float] = {}
        train_up, train_down, train_mean = [], [], []

        for idx in train_idx:
            pattern_id = int(pattern_df.at[idx, col])
            n_seen = counts.get(pattern_id, 0)
            if n_seen >= MIN_PATTERN_OBS:
                train_up.append(up_counts.get(pattern_id, 0) / n_seen)
                train_down.append(down_counts.get(pattern_id, 0) / n_seen)
                train_mean.append(sums.get(pattern_id, 0.0) / n_seen)
            else:
                train_up.append(default_up)
                train_down.append(default_down)
                train_mean.append(default_mean)

            label = labels[idx]
            if not np.isnan(label):
                label = int(label)
                counts[pattern_id] = n_seen + 1
                up_counts[pattern_id] = up_counts.get(pattern_id, 0) + int(label == 1)
                down_counts[pattern_id] = down_counts.get(pattern_id, 0) + int(label == -1)
                sums[pattern_id] = sums.get(pattern_id, 0.0) + label

        def lookup(pattern_id: int, kind: str) -> float:
            n_seen = counts.get(pattern_id, 0)
            if n_seen < MIN_PATTERN_OBS:
                if kind == "up":
                    return default_up
                if kind == "down":
                    return default_down
                return default_mean
            if kind == "up":
                return up_counts.get(pattern_id, 0) / n_seen
            if kind == "down":
                return down_counts.get(pattern_id, 0) / n_seen
            return sums.get(pattern_id, 0.0) / n_seen

        train_stats[f"{col}_up_rate"] = train_up
        train_stats[f"{col}_down_rate"] = train_down
        train_stats[f"{col}_label_mean"] = train_mean
        test_patterns = pattern_df.loc[test_idx, col].astype(int)
        test_stats[f"{col}_up_rate"] = [lookup(pid, "up") for pid in test_patterns]
        test_stats[f"{col}_down_rate"] = [lookup(pid, "down") for pid in test_patterns]
        test_stats[f"{col}_label_mean"] = [lookup(pid, "mean") for pid in test_patterns]

    return train_stats.fillna(0.0), test_stats.fillna(0.0)


def walk_forward_pattern_features(
    feat_df: pd.DataFrame,
    pattern_df: pd.DataFrame,
    labels: np.ndarray,
    baseline_cols: list[str],
) -> tuple[dict, dict]:
    base_avail = [col for col in baseline_cols if col in feat_df.columns]
    y_true_all, y_pred_base_all, y_pred_candidate_all = [], [], []
    n = len(feat_df)

    cutoff = MIN_TRAIN
    while cutoff + STEP + LABEL_HORIZON <= n:
        train_end = cutoff - LABEL_HORIZON
        if train_end < MIN_TRAIN:
            cutoff += STEP
            continue

        y_train_full = labels[:train_end]
        valid_train = np.where(~np.isnan(y_train_full))[0]
        if len(valid_train) < MIN_TRAIN or len(np.unique(y_train_full[valid_train])) < 2:
            cutoff += STEP
            continue

        test_end = min(cutoff + STEP, n - LABEL_HORIZON)
        y_test = labels[cutoff:test_end]
        valid_test_mask = ~np.isnan(y_test)
        if valid_test_mask.sum() == 0:
            cutoff += STEP
            continue
        test_idx = np.arange(cutoff, test_end)[valid_test_mask]

        X_train_base = feat_df.loc[valid_train, base_avail].fillna(0.0)
        X_test_base = feat_df.loc[test_idx, base_avail].fillna(0.0)

        train_stats, test_stats = _pattern_stats_for_fold(pattern_df, labels, valid_train, test_idx)
        X_train_candidate = pd.concat(
            [
                X_train_base.reset_index(drop=True),
                pattern_df.loc[valid_train, RAW_PATTERN_FEATURES].reset_index(drop=True),
                train_stats.reset_index(drop=True),
            ],
            axis=1,
        )
        X_test_candidate = pd.concat(
            [
                X_test_base.reset_index(drop=True),
                pattern_df.loc[test_idx, RAW_PATTERN_FEATURES].reset_index(drop=True),
                test_stats.reset_index(drop=True),
            ],
            axis=1,
        )

        y_train = y_train_full[valid_train].astype(int)
        base_model = RandomForestClassifier(**RF_PARAMS)
        candidate_model = RandomForestClassifier(**RF_PARAMS)
        base_model.fit(X_train_base.values, y_train)
        candidate_model.fit(X_train_candidate.values, y_train)

        y_pred_base = base_model.predict(X_test_base.values)
        y_pred_candidate = candidate_model.predict(X_test_candidate.values)

        y_true_all.extend(y_test[valid_test_mask].tolist())
        y_pred_base_all.extend(y_pred_base.tolist())
        y_pred_candidate_all.extend(y_pred_candidate.tolist())
        cutoff += STEP

    if not y_true_all:
        return {}, {}

    y_true = np.array(y_true_all)
    baseline = compute_metrics(y_true, np.array(y_pred_base_all))
    candidate = compute_metrics(y_true, np.array(y_pred_candidate_all))
    candidate["added_features"] = len(RAW_PATTERN_FEATURES) + len(PATTERN_ID_COLUMNS) * 3
    return baseline, candidate


def _mean(rows: list[dict], field: str) -> float:
    vals = [
        row[field]
        for row in rows
        if isinstance(row.get(field), (int, float)) and not np.isnan(row.get(field, float("nan")))
    ]
    return round(sum(vals) / len(vals), 1) if vals else float("nan")


def _aggregate(rows: list[dict]) -> dict:
    return {
        "accuracy": _mean(rows, "accuracy"),
        "dir_accuracy": _mean(rows, "dir_accuracy"),
        "up_precision": _mean(rows, "up_precision"),
        "dn_precision": _mean(rows, "dn_precision"),
        "n_signals": _mean(rows, "n_signals"),
        "n_predictions": _mean(rows, "n_predictions"),
    }


def _json_default(value):
    if isinstance(value, np.integer):
        return int(value)
    if isinstance(value, np.floating):
        return float(value)
    if isinstance(value, np.bool_):
        return bool(value)
    raise TypeError(f"Object of type {type(value).__name__} is not JSON serializable")


def run(stocks: list[str], output_path: Path) -> dict:
    results = {}
    agg_base, agg_candidate = [], []
    baseline_cols = list(FEATURE_COLUMNS)

    hdr = f"{'Stock':>6}  {'Model':>12}  {'Acc%':>5}  {'Dir%':>5}  {'Up%':>6}  {'Signals':>7}"
    print(f"\n{hdr}\n{'-' * len(hdr)}")
    for stock_no in stocks:
        print(f"  computing {stock_no}...", end="\r", flush=True)
        df = fetch_validation_df(stock_no)
        if df is None or df.empty:
            print(f"{stock_no:>6}  no data")
            continue

        feat = _build_features(df)
        pattern = build_pattern_features(df)
        close = (df.set_index("date")["close"] if "date" in df.columns else df["close"]).values
        labels = build_triple_barrier_labels(close)
        baseline, candidate = walk_forward_pattern_features(feat, pattern, labels, baseline_cols)
        if not baseline:
            print(f"{stock_no:>6}  insufficient folds")
            continue

        acc_delta = round(candidate["accuracy"] - baseline["accuracy"], 1)
        passed = acc_delta >= PASS_PER_STOCK_ACCURACY_DELTA
        results[stock_no] = {
            "baseline": baseline,
            "pattern_mining": candidate,
            "accuracy_delta_pp": acc_delta,
            "passed": passed,
        }
        agg_base.append(baseline)
        agg_candidate.append(candidate)

        for name, row in (("baseline", baseline), ("c19_pattern", candidate)):
            prefix = f"{stock_no:>6}" if name == "baseline" else f"{'':>6}"
            print(
                f"{prefix}  {name:>12}  {row['accuracy']:>5.1f}  {row['dir_accuracy']:>5.1f}  "
                f"{row['up_precision']:>6.1f}  {row['n_signals']:>7}"
            )
        print(f"{'':>6}  {'delta':>12}  {acc_delta:>+5.1f}  {'PASS' if passed else 'FAIL'}")

    aggregate = {"baseline": _aggregate(agg_base), "pattern_mining": _aggregate(agg_candidate)}
    per_stock_passed = bool(results) and all(row["passed"] for row in results.values())
    acc_delta = round(aggregate["pattern_mining"]["accuracy"] - aggregate["baseline"]["accuracy"], 1)
    signal_delta = round(aggregate["pattern_mining"]["n_signals"] - aggregate["baseline"]["n_signals"], 1)

    print("\n=== AGGREGATE ===")
    for name, row in aggregate.items():
        print(
            f"  {name:>15}: acc={row['accuracy']}%  dir={row['dir_accuracy']}%  "
            f"up={row['up_precision']}%  signals={row['n_signals']}"
        )
    print(f"\n  Delta accuracy: {acc_delta:+.1f}pp")
    print(f"  Delta signals:  {signal_delta:+.1f} per stock")
    print(
        f"  Pass every stock +{PASS_PER_STOCK_ACCURACY_DELTA:.1f}pp: "
        f"{'YES' if per_stock_passed else 'NO'}"
    )

    result = {
        "experiment": "C19_pattern_mining",
        "description": "No-lookahead dry run for candlestick, SAX, and fold-local pattern-stat features.",
        "stocks": stocks,
        "pass_criterion": {
            "every_stock_accuracy_delta_pp_at_least": PASS_PER_STOCK_ACCURACY_DELTA,
        },
        "baseline_features": baseline_cols,
        "candidate_raw_pattern_features": RAW_PATTERN_FEATURES,
        "candidate_fold_local_pattern_stats": [
            f"{col}_{suffix}"
            for col in PATTERN_ID_COLUMNS
            for suffix in ("up_rate", "down_rate", "label_mean")
        ],
        "aggregate": aggregate,
        "aggregate_accuracy_delta_pp": acc_delta,
        "aggregate_signal_delta": signal_delta,
        "per_stock": results,
        "passed": per_stock_passed,
        "promotion_decision": "integrate" if per_stock_passed else "do_not_integrate",
        "validation": {
            "mode": "walk_forward_with_embargo",
            "label_horizon": LABEL_HORIZON,
            "train_end": "cutoff - LABEL_HORIZON",
            "step": STEP,
            "min_train": MIN_TRAIN,
            "leakage_guard": "pattern stats for test rows are learned only from the training slice; training rows use prior expanding stats.",
        },
    }

    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps(result, indent=2, ensure_ascii=False, default=_json_default))
    print(f"\nSaved -> {output_path}")
    return result


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--stocks", default="", help="Comma-separated stock list")
    parser.add_argument("--extended", action="store_true", help="Use 12-stock extended validation")
    parser.add_argument("--output", default="docs/c19_pattern_mining_result.json")
    args = parser.parse_args()

    if args.stocks:
        stocks = [code.strip() for code in args.stocks.split(",") if code.strip()]
    else:
        stocks = EXTENDED_STOCKS if args.extended else DEFAULT_STOCKS

    result = run(stocks, ROOT / args.output)
    return 0 if result.get("passed") else 1


if __name__ == "__main__":
    raise SystemExit(main())