#!/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())