865 lines
52 KiB
Python
865 lines
52 KiB
Python
"""Command-line entry points for download, market, scan, backtest and research."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import sys
|
|
import time
|
|
import warnings
|
|
|
|
import pandas as pd
|
|
|
|
from .backtest import run_backtest
|
|
from .config import AppConfig
|
|
from .corporate_actions import (
|
|
SiamChartActionError,
|
|
audit_siamchart_actions,
|
|
fetch_set_cash_dividend_reference,
|
|
fetch_siamchart_stock_info,
|
|
merge_siamchart_dividends,
|
|
read_siamchart_dividend_sidecar,
|
|
)
|
|
from .data import SymbolDataError, YFinanceProvider, normalize_bars, read_siamchart_archives, safe_filename
|
|
from .ema_crossover import run_ema_crossover_backtest
|
|
from .experiments import run_ablations, run_chronological_holdouts, rolling_walk_forward
|
|
from .model_lab import run_model_lab
|
|
from .research import build_feature_panel, build_market_context
|
|
from .scanner import scan_candidates
|
|
from .synthetic import SMOKE_SYMBOLS, synthetic_bars
|
|
from .universe import MembershipError, PointInTimeUniverse, resolve_universe
|
|
|
|
|
|
DEFAULT_MEMBERSHIP = Path("data/universes/set100_2026_h1.csv")
|
|
|
|
|
|
def _parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(prog="python -m primethai", description="Thai stock research and backtesting; no broker execution.")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
download = sub.add_parser("download", help="download/cache daily Yahoo Finance bars")
|
|
download.add_argument("--symbols", nargs="*", help="symbols without .BK; defaults to the 10-symbol smoke list")
|
|
download.add_argument("--universe", default="custom", choices=["SET100", "SET50", "SET", "mai", "custom"])
|
|
download.add_argument("--membership-file", default=str(DEFAULT_MEMBERSHIP))
|
|
download.add_argument("--mode", choices=["fast", "accurate"], default="fast")
|
|
download.add_argument("--start")
|
|
download.add_argument("--end")
|
|
download.add_argument("--period", default="5y")
|
|
download.add_argument("--refresh", action="store_true")
|
|
download.add_argument("--data-dir", default="data/raw")
|
|
|
|
siamchart = sub.add_parser("import-siamchart", help="import selected tickers from SiamChart EOD ZIP archives")
|
|
siamchart.add_argument("--archives", nargs="+", required=True, help="SiamChart set-archive_EOD ZIP files")
|
|
siamchart.add_argument("--symbols", nargs="+", default=["^SET"], help="tickers to extract; defaults to the SET index")
|
|
siamchart.add_argument("--data-dir", default="data/raw")
|
|
|
|
actions = sub.add_parser("import-siamchart-actions", help="import dividend and par-value history from SiamChart stock-info pages")
|
|
actions.add_argument("--symbols", nargs="+", required=True, help="tickers without .BK")
|
|
actions.add_argument("--data-dir", default="data/raw")
|
|
actions.add_argument("--delay-seconds", type=float, default=0.5,
|
|
help="pause between stock-info requests to avoid excessive traffic")
|
|
|
|
verify_actions = sub.add_parser("verify-siamchart-actions", help="compare action sidecars with a SET reference CSV")
|
|
verify_actions.add_argument("--symbols", nargs="+", required=True, help="tickers without .BK")
|
|
verify_actions.add_argument("--set-reference", required=True,
|
|
help="CSV with symbol,action_type,date,value and optional source_url columns")
|
|
verify_actions.add_argument("--data-dir", default="data/raw")
|
|
verify_actions.add_argument("--output", default="reports/siamchart_actions_audit.csv")
|
|
verify_actions.add_argument("--start", help="inclusive action date")
|
|
verify_actions.add_argument("--end", help="exclusive action date")
|
|
verify_actions.add_argument("--action-types", nargs="+", choices=["dividend", "par_value"],
|
|
default=["dividend"], help="action classes to compare; defaults to dividends")
|
|
|
|
set_reference = sub.add_parser("import-set-actions", help="import cash-dividend records from SET's public rights-and-benefits history")
|
|
set_reference.add_argument("--symbols", nargs="+", required=True, help="tickers without .BK")
|
|
set_reference.add_argument("--start", default="2025-01-01", help="inclusive action date")
|
|
set_reference.add_argument("--end", default="2026-07-01", help="exclusive action date")
|
|
set_reference.add_argument("--output", default="data/reference/set_actions_2025_2026_h1.csv")
|
|
set_reference.add_argument("--delay-seconds", type=float, default=0.35)
|
|
set_reference.add_argument("--symbol-aliases", default="data/reference/set_symbol_aliases.csv",
|
|
help="optional CSV mapping SET action symbols to the historical price-series symbol")
|
|
|
|
model_lab = sub.add_parser("model-lab", help="screen long-history Thai stock strategy candidates")
|
|
model_lab.add_argument("--archives", nargs="+", default=[
|
|
"data/raw/siamchart/set-archive_EOD_1970-LAST.zip",
|
|
"data/raw/siamchart/set-archive_EOD.zip",
|
|
], help="SiamChart market-wide EOD ZIP archives")
|
|
model_lab.add_argument("--start", default="2005-01-01", help="inclusive evaluation start")
|
|
model_lab.add_argument("--end", default="2026-07-01", help="exclusive evaluation end")
|
|
model_lab.add_argument("--output-dir", default="reports/model_lab")
|
|
model_lab.add_argument("--action-dir", default="data/raw/actions")
|
|
model_lab.add_argument("--set-index-csv", default="data/raw/^SET.csv")
|
|
model_lab.add_argument("--set-reference", default="data/reference/set_actions_2025_2026_h1.csv")
|
|
model_lab.add_argument("--initial-equity", type=float, default=1_000_000.0)
|
|
|
|
webui = sub.add_parser("webui", help="run the portfolio app with automatic SiamChart updates")
|
|
webui.add_argument("--host", default="127.0.0.1", help="bind address; remote use requires an HTTPS reverse proxy")
|
|
webui.add_argument("--port", type=int, default=18900)
|
|
webui.add_argument("--archives", nargs="+", default=[
|
|
"data/raw/siamchart/set-archive_EOD_1970-LAST.zip",
|
|
"data/raw/siamchart/set-archive_EOD.zip",
|
|
"data/raw/siamchart/set-archive_EOD_LAST.zip",
|
|
], help="SiamChart market-wide EOD ZIP archives")
|
|
webui.add_argument("--start", default="2026-01-01", help="fresh-start portfolio evaluation start")
|
|
webui.add_argument("--initial-equity", type=float, default=1_000_000.0)
|
|
webui.add_argument("--report-dir", default="reports/primary_union_current")
|
|
|
|
for command in ("market", "scan", "backtest", "experiment", "ema-crossover", "strategy-overlays"):
|
|
item = sub.add_parser(command)
|
|
item.add_argument("--universe", default="SET100", choices=["SET100", "SET50", "SET", "mai", "custom"])
|
|
item.add_argument("--membership-file", default=str(DEFAULT_MEMBERSHIP))
|
|
item.add_argument("--mode", choices=["fast", "accurate"], default="fast")
|
|
item.add_argument("--symbols", nargs="*", help="symbols for custom universe")
|
|
item.add_argument("--start")
|
|
item.add_argument("--end")
|
|
item.add_argument("--data-dir", default="data/raw")
|
|
item.add_argument("--config", default="config/default.json")
|
|
item.add_argument("--output-dir", default="reports")
|
|
item.add_argument("--set-reference", help="complete SET cash-dividend reference CSV from import-set-actions")
|
|
if command == "scan":
|
|
item.add_argument("--top", type=int, default=10, choices=[5, 10, 20])
|
|
item.add_argument("--json-out")
|
|
if command == "ema-crossover":
|
|
item.add_argument("--ema-pairs", nargs="+", default=["10:30", "20:50", "50:200"],
|
|
help="fast:slow EMA spans to compare; default: 10:30 20:50 50:200")
|
|
item.add_argument("--max-position-pct", type=float,
|
|
help="target portfolio fraction per position; defaults to config")
|
|
item.add_argument("--min-price", type=float,
|
|
help="minimum signal close; defaults to config")
|
|
item.add_argument("--min-value-turnover-thb", type=float,
|
|
help="minimum 20-day average value turnover; defaults to config")
|
|
item.add_argument("--market-filters", nargs="+",
|
|
choices=["none", "ew_trend_up", "ew_above_ema200"], default=["none"],
|
|
help="entry gates to compare")
|
|
item.add_argument("--exit-modes", nargs="+", choices=["cross_under", "ema21_low"],
|
|
default=["cross_under"], help="exit rules to compare")
|
|
item.add_argument("--max-loss-pcts", nargs="+", type=float, default=[0.0],
|
|
help="hard intraday loss stops to compare; 0 disables, e.g. 0.08 0.12")
|
|
item.add_argument("--split-by-year", action="store_true",
|
|
help="also run independent simulations for each calendar-year slice")
|
|
sub.add_parser("smoke", help="run the full research pipeline on synthetic offline data")
|
|
return parser
|
|
|
|
|
|
def _membership(path: str | None, universe: str) -> PointInTimeUniverse | None:
|
|
if universe.lower() == "custom":
|
|
return None
|
|
if not path or not Path(path).exists():
|
|
raise MembershipError(f"membership CSV not found: {path}. Use --membership-file with dated constituents.")
|
|
return PointInTimeUniverse.from_csv(path, name=universe.upper())
|
|
|
|
|
|
def _symbols_for_download(args: argparse.Namespace) -> tuple[list[str], PointInTimeUniverse | None]:
|
|
membership = _membership(args.membership_file, args.universe) if args.universe.lower() != "custom" else None
|
|
if args.symbols:
|
|
symbols = [symbol.upper().removesuffix(".BK") for symbol in args.symbols]
|
|
elif membership:
|
|
symbols = sorted({row.symbol for row in membership.periods})
|
|
else:
|
|
symbols = SMOKE_SYMBOLS.copy()
|
|
return sorted(set(symbols + ["^SET"])), membership
|
|
|
|
|
|
def _read_bars(directory: str | Path, symbol: str) -> pd.DataFrame | None:
|
|
directory = Path(directory)
|
|
options = [directory / f"{safe_filename(symbol)}.csv"]
|
|
if symbol == "^SET":
|
|
options += [directory / "^SET.csv", directory / "^SET.BK.csv", directory / "SET.csv"]
|
|
else:
|
|
options += [directory / f"{safe_filename(symbol + '.BK')}.csv"]
|
|
for path in options:
|
|
if path.exists():
|
|
frame = pd.read_csv(path)
|
|
date_column = next((column for column in frame.columns
|
|
if str(column).strip().lower() == "date"), None)
|
|
if date_column is None:
|
|
raise SymbolDataError(f"CSV has no date column: {path}")
|
|
frame = frame.set_index(date_column)
|
|
if symbol == "^SET" and not any(str(column).strip().lower() == "volume"
|
|
for column in frame.columns):
|
|
frame["volume"] = 0.0
|
|
bars = normalize_bars(frame)
|
|
action_path = directory / "actions" / f"{safe_filename(symbol)}_dividends.csv"
|
|
if action_path.exists():
|
|
dividends = read_siamchart_dividend_sidecar(action_path, symbol)
|
|
bars = merge_siamchart_dividends(bars, dividends, symbol)
|
|
return bars
|
|
return None
|
|
|
|
|
|
def _accurate_context_bars(set_bars: pd.DataFrame, membership: PointInTimeUniverse,
|
|
start: str | None, end: str | None) -> pd.DataFrame:
|
|
"""Keep PIT-covered market history through the evaluation end, including warm-up."""
|
|
coverage_start = min(row.start for row in membership.periods)
|
|
coverage_end = max(row.end for row in membership.periods)
|
|
evaluation_start = pd.Timestamp(start) if start else coverage_start
|
|
evaluation_end = pd.Timestamp(end) if end else coverage_end + pd.Timedelta(days=1)
|
|
if evaluation_start < coverage_start or evaluation_start > coverage_end:
|
|
raise MembershipError(
|
|
f"requested backtest start {evaluation_start.date()} is outside {membership.name} PIT coverage "
|
|
f"({coverage_start.date()} to {coverage_end.date()})"
|
|
)
|
|
if evaluation_end <= evaluation_start:
|
|
raise ValueError("--end must be later than --start")
|
|
if evaluation_end > coverage_end + pd.Timedelta(days=1):
|
|
raise MembershipError(
|
|
f"requested backtest end {evaluation_end.date()} exceeds {membership.name} PIT coverage "
|
|
f"through {coverage_end.date()}"
|
|
)
|
|
clipped = set_bars.loc[(set_bars.index >= coverage_start) & (set_bars.index < evaluation_end)]
|
|
if clipped.empty:
|
|
raise SymbolDataError("no SET sessions fall inside the requested PIT coverage and date range")
|
|
if len(clipped) < 260:
|
|
raise SymbolDataError(
|
|
f"only {len(clipped)} SET sessions remain inside PIT coverage and the requested end; "
|
|
"at least 260 are needed for the 52-week market warm-up"
|
|
)
|
|
membership.require_coverage(clipped.index)
|
|
return clipped
|
|
|
|
|
|
def _prepare(args: argparse.Namespace, config: AppConfig) -> tuple[dict[str, pd.DataFrame], pd.DataFrame, pd.DataFrame, PointInTimeUniverse | None, list[str]]:
|
|
membership = _membership(args.membership_file, args.universe) if args.universe.lower() != "custom" else None
|
|
set_bars = _read_bars(args.data_dir, "^SET")
|
|
if set_bars is None:
|
|
raise SymbolDataError(
|
|
"^SET.BK daily history is missing or incomplete. A provider may have returned only a current quote; "
|
|
"supply a full daily OHLCV CSV as data/raw/^SET.csv or data/raw/^SET.BK.csv, or use a provider "
|
|
"with historical SET index data"
|
|
)
|
|
if len(set_bars) < 260:
|
|
raise SymbolDataError(
|
|
f"^SET.BK has only {len(set_bars)} daily bars; at least 260 are needed for the 52-week "
|
|
"market warm-up. Supply a longer index history before scanning or backtesting"
|
|
)
|
|
if membership:
|
|
symbols = sorted({row.symbol for row in membership.periods})
|
|
else:
|
|
if not args.symbols:
|
|
raise ValueError("--universe custom requires --symbols")
|
|
symbols = sorted({symbol.upper().removesuffix(".BK") for symbol in args.symbols})
|
|
stock_bars = {symbol: bars for symbol in symbols if (bars := _read_bars(args.data_dir, symbol)) is not None}
|
|
external_dividends: dict[str, dict[pd.Timestamp, float]] = {}
|
|
for symbol, bars in stock_bars.items():
|
|
events = bars.attrs.get("external_dividends", {})
|
|
if events:
|
|
external_dividends[symbol] = {
|
|
pd.Timestamp(date).normalize(): float(amount) for date, amount in events.items()
|
|
}
|
|
if args.set_reference:
|
|
reference_path = Path(args.set_reference)
|
|
manifest_path = Path(str(reference_path) + ".manifest.json")
|
|
if not reference_path.exists() or not manifest_path.exists():
|
|
raise SiamChartActionError(
|
|
f"SET backtest reference requires both {reference_path} and its coverage manifest {manifest_path}; "
|
|
"create them with import-set-actions"
|
|
)
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
if not manifest.get("coverage_complete"):
|
|
raise SiamChartActionError("SET reference coverage is incomplete; refusing an action-adjusted backtest")
|
|
covered_symbols = {str(symbol).upper() for symbol in manifest.get("requested_price_symbols", [])}
|
|
uncovered_symbols = sorted(set(symbols) - covered_symbols)
|
|
if uncovered_symbols:
|
|
raise SiamChartActionError(
|
|
"SET reference does not cover selected price symbols: " + ", ".join(uncovered_symbols[:20])
|
|
)
|
|
coverage_start = pd.Timestamp(manifest["start_inclusive"]).normalize()
|
|
coverage_end = pd.Timestamp(manifest["end_exclusive"]).normalize()
|
|
reference = pd.read_csv(reference_path, dtype={"symbol": str, "action_type": str})
|
|
required = {"symbol", "action_type", "date", "value"}
|
|
if not required.issubset(reference.columns):
|
|
raise SiamChartActionError("SET reference CSV must contain symbol, action_type, date and value")
|
|
reference["symbol"] = reference["symbol"].str.upper().str.strip()
|
|
reference["action_type"] = reference["action_type"].str.lower().str.strip()
|
|
reference["date"] = pd.to_datetime(reference["date"], errors="coerce").dt.normalize()
|
|
reference["value"] = pd.to_numeric(reference["value"], errors="coerce")
|
|
if reference[["date", "value"]].isna().any().any():
|
|
raise SiamChartActionError("SET reference CSV contains invalid dates or dividend values")
|
|
reference = reference.loc[
|
|
reference["symbol"].isin(symbols)
|
|
& (reference["action_type"] == "dividend")
|
|
& (reference["date"] >= coverage_start)
|
|
& (reference["date"] < coverage_end)
|
|
]
|
|
if reference.duplicated(["symbol", "date"]).any():
|
|
raise SiamChartActionError("SET reference has duplicate cash-dividend rows for a symbol and date")
|
|
excluded: list[tuple[str, pd.Timestamp, float]] = []
|
|
value_differences: list[tuple[str, pd.Timestamp, float, float]] = []
|
|
for symbol, bars in stock_bars.items():
|
|
events = external_dividends.setdefault(symbol, {})
|
|
set_events = {
|
|
pd.Timestamp(row.date).normalize(): float(row.value)
|
|
for row in reference.loc[reference["symbol"] == symbol].itertuples(index=False)
|
|
}
|
|
for date, amount in list(events.items()):
|
|
if coverage_start <= date < coverage_end:
|
|
if date not in set_events:
|
|
excluded.append((symbol, date, amount))
|
|
elif abs(amount - set_events[date]) > 5.1e-7:
|
|
value_differences.append((symbol, date, amount, set_events[date]))
|
|
del events[date]
|
|
events.update(set_events)
|
|
# Suppress provider-native dividend values inside the covered period;
|
|
# the complete SET event list is the only cash-action source there.
|
|
in_coverage = (bars.index >= coverage_start) & (bars.index < coverage_end)
|
|
if "dividends" in bars.columns and in_coverage.any():
|
|
bars.loc[in_coverage, "dividends"] = 0.0
|
|
if excluded:
|
|
detail = ", ".join(f"{symbol} {date.date()} {amount:g}" for symbol, date, amount in excluded[:12])
|
|
warnings.warn(
|
|
f"SET is authoritative for cash dividends from {coverage_start.date()} through "
|
|
f"{(coverage_end - pd.Timedelta(days=1)).date()}; excluded SiamChart-only events pending review: {detail}",
|
|
RuntimeWarning,
|
|
)
|
|
if value_differences:
|
|
detail = ", ".join(f"{symbol} {date.date()} SiamChart={old:g} SET={new:g}"
|
|
for symbol, date, old, new in value_differences[:12])
|
|
warnings.warn("SET amounts replaced differing SiamChart amounts: " + detail, RuntimeWarning)
|
|
print(f"SET cash actions applied: {len(reference)} event(s), coverage {coverage_start.date()} to "
|
|
f"{(coverage_end - pd.Timedelta(days=1)).date()} inclusive.")
|
|
missing = sorted(set(symbols) - set(stock_bars))
|
|
if missing:
|
|
print("No cached data for: " + ", ".join(missing), file=sys.stderr)
|
|
if args.mode == "accurate":
|
|
raise SymbolDataError(f"accurate membership mode is missing historical OHLCV for {len(missing)} constituents; refusing incomplete breadth/backtest")
|
|
print("SURVIVORSHIP / COVERAGE WARNING: breadth and SET100EW will use only downloaded members.", file=sys.stderr)
|
|
if not stock_bars:
|
|
raise SymbolDataError("no constituent price data available")
|
|
if args.mode == "accurate":
|
|
if membership is None:
|
|
raise MembershipError("accurate mode requires a dated membership file")
|
|
set_bars = _accurate_context_bars(set_bars, membership, args.start, args.end)
|
|
session_dates = set_bars.index
|
|
stock_bars = {symbol: bars.loc[bars.index.isin(session_dates)] for symbol, bars in stock_bars.items()}
|
|
market, _, _ = build_market_context(stock_bars, set_bars, membership, accurate=True,
|
|
research=config.research)
|
|
warnings.warn("ACCURATE membership mode uses the supplied PIT file. Price history and corporate actions depend on the selected source and must be validated before performance claims.", RuntimeWarning)
|
|
else:
|
|
dates = set_bars.index
|
|
members, _ = resolve_universe(args.universe, membership_file=args.membership_file if membership else None,
|
|
symbols=args.symbols, mode="fast", dates=dates)
|
|
market, _, _ = build_market_context(stock_bars, set_bars, membership,
|
|
current_members=members, accurate=False,
|
|
research=config.research)
|
|
market.attrs["external_dividends"] = external_dividends
|
|
panel = build_feature_panel(stock_bars, set_bars, market, rs_weights=config.rs_weights,
|
|
volume_ratio=config.strategy.volume_pullback_ratio,
|
|
research=config.research)
|
|
return panel, market, set_bars, membership, symbols
|
|
|
|
|
|
def _load_config(path: str | None) -> AppConfig:
|
|
return AppConfig.load(path) if path and Path(path).exists() else AppConfig()
|
|
|
|
|
|
def _parse_ema_pair(value: str) -> tuple[int, int]:
|
|
try:
|
|
fast_text, slow_text = value.split(":", 1)
|
|
fast, slow = int(fast_text), int(slow_text)
|
|
except (ValueError, TypeError) as exc:
|
|
raise ValueError(f"invalid EMA pair {value!r}; use FAST:SLOW such as 20:50") from exc
|
|
if fast < 1 or slow < 2 or fast >= slow:
|
|
raise ValueError(f"invalid EMA pair {value!r}; require positive spans with fast < slow")
|
|
return fast, slow
|
|
|
|
|
|
def _ema_periods(args: argparse.Namespace, market: pd.DataFrame) -> list[tuple[str, str, str]]:
|
|
start = pd.Timestamp(args.start).normalize() if args.start else pd.Timestamp(market.index.min()).normalize()
|
|
end = pd.Timestamp(args.end).normalize() if args.end else pd.Timestamp(market.index.max()).normalize() + pd.Timedelta(days=1)
|
|
if end <= start:
|
|
raise ValueError("--end must be later than --start")
|
|
periods = [("full", start.date().isoformat(), end.date().isoformat())]
|
|
if args.split_by_year:
|
|
cursor = start
|
|
while cursor < end:
|
|
next_year = pd.Timestamp(year=cursor.year + 1, month=1, day=1)
|
|
period_end = min(next_year, end)
|
|
if cursor.month == 1 and cursor.day == 1 and period_end == next_year:
|
|
label = str(cursor.year)
|
|
else:
|
|
label = f"{cursor.date().isoformat()}_to_{(period_end - pd.Timedelta(days=1)).date().isoformat()}"
|
|
periods.append((label, cursor.date().isoformat(), period_end.date().isoformat()))
|
|
cursor = period_end
|
|
return periods
|
|
|
|
|
|
def _download(args: argparse.Namespace) -> int:
|
|
symbols, membership = _symbols_for_download(args)
|
|
if args.mode == "accurate":
|
|
if membership is None or not args.start:
|
|
raise MembershipError("accurate downloads require a dated membership file and explicit --start/--end")
|
|
end = pd.Timestamp(args.end) - pd.Timedelta(days=1) if args.end else pd.Timestamp.now().normalize()
|
|
membership.require_coverage(pd.bdate_range(pd.Timestamp(args.start), end))
|
|
print(f"PIT membership validation passed for {args.universe} from {args.start} through {end.date()}.")
|
|
elif membership is None:
|
|
print("SURVIVORSHIP BIAS: current/custom sample symbols are for pipeline development only.")
|
|
else:
|
|
latest_end = max(row.end for row in membership.periods)
|
|
print(f"SURVIVORSHIP BIAS: fast download uses the supplied constituent snapshot (latest effective date {latest_end.date()}) for research only.")
|
|
provider = YFinanceProvider()
|
|
data, errors = provider.download_many(symbols, start=args.start, end=args.end,
|
|
period=args.period if not args.start else None,
|
|
refresh=args.refresh)
|
|
output = Path(args.data_dir)
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
for symbol, frame in data.items():
|
|
destination = output / f"{safe_filename(symbol)}.csv"
|
|
frame.to_csv(destination, index_label="date")
|
|
print(f"{symbol}: {len(frame)} daily bars -> {destination}")
|
|
if errors:
|
|
print("\nDownload issues (other symbols were retained):", file=sys.stderr)
|
|
for symbol, message in errors.items():
|
|
print(f"- {symbol}: {message}", file=sys.stderr)
|
|
print(f"\nDownloaded {len(data)}/{len(symbols)} symbols. auto_adjust=False; Adj Close, dividends and splits retained.")
|
|
return 1 if not data else 0
|
|
|
|
|
|
def _import_siamchart(args: argparse.Namespace) -> int:
|
|
data, issues = read_siamchart_archives(args.archives, args.symbols)
|
|
output = Path(args.data_dir)
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
for symbol, frame in sorted(data.items()):
|
|
filename = "^SET.csv" if symbol == "^SET" else f"{safe_filename(symbol)}.csv"
|
|
destination = output / filename
|
|
frame.to_csv(destination, index_label="date")
|
|
print(f"{symbol}: {len(frame)} daily bars ({frame.index.min().date()} to {frame.index.max().date()}) -> {destination}")
|
|
if issues:
|
|
print(f"\nSiamChart entries skipped: {len(issues)}", file=sys.stderr)
|
|
for issue in issues[:20]:
|
|
print(f"- {issue}", file=sys.stderr)
|
|
if len(issues) > 20:
|
|
print(f"- ... and {len(issues) - 20} more", file=sys.stderr)
|
|
print("SiamChart EOD archives contain raw OHLCV. Separate stock-info pages list dividend and par-value history, but this importer does not ingest those tables.")
|
|
return 0
|
|
|
|
|
|
def _import_siamchart_actions(args: argparse.Namespace) -> int:
|
|
if args.delay_seconds < 0:
|
|
raise ValueError("--delay-seconds must be zero or greater")
|
|
symbols = sorted({symbol.strip().upper().removesuffix(".BK") for symbol in args.symbols if symbol.strip()})
|
|
if not symbols:
|
|
raise ValueError("select at least one ticker to import")
|
|
action_dir = Path(args.data_dir) / "actions"
|
|
action_dir.mkdir(parents=True, exist_ok=True)
|
|
failures: list[str] = []
|
|
succeeded = 0
|
|
for index, symbol in enumerate(symbols):
|
|
if index and args.delay_seconds:
|
|
time.sleep(args.delay_seconds)
|
|
try:
|
|
actions = fetch_siamchart_stock_info(symbol)
|
|
except SiamChartActionError as exc:
|
|
failures.append(str(exc))
|
|
continue
|
|
dividend_path = action_dir / f"{safe_filename(symbol)}_dividends.csv"
|
|
par_path = action_dir / f"{safe_filename(symbol)}_par_changes.csv"
|
|
actions.dividends.to_csv(dividend_path, index=False)
|
|
actions.par_changes.to_csv(par_path, index=False)
|
|
print(f"{symbol}: {len(actions.dividends)} dividend event(s), {len(actions.par_changes)} par-value row(s)")
|
|
print(f" -> {dividend_path}")
|
|
print(f" -> {par_path}")
|
|
succeeded += 1
|
|
if failures:
|
|
print("\nSiamChart stock-info pages not imported:", file=sys.stderr)
|
|
for failure in failures:
|
|
print(f"- {failure}", file=sys.stderr)
|
|
print(f"\nImported {succeeded}/{len(symbols)} tickers. Dividend dates are treated as XD dates; par-value changes remain review-only.")
|
|
return 1 if failures or not succeeded else 0
|
|
|
|
|
|
def _verify_siamchart_actions(args: argparse.Namespace) -> int:
|
|
symbols = sorted({symbol.strip().upper().removesuffix(".BK") for symbol in args.symbols if symbol.strip()})
|
|
if not symbols:
|
|
raise ValueError("select at least one ticker to verify")
|
|
result = audit_siamchart_actions(Path(args.data_dir) / "actions", args.set_reference, symbols,
|
|
start=args.start, end=args.end,
|
|
action_types=args.action_types)
|
|
output = Path(args.output)
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
result.to_csv(output, index=False)
|
|
counts = result["status"].value_counts().to_dict() if not result.empty else {}
|
|
print("SET versus SiamChart action audit")
|
|
for status in ("MATCH", "VALUE_MISMATCH", "ONLY_SET", "ONLY_SIAMCHART"):
|
|
if status in counts:
|
|
print(f"{status:18s} {counts[status]}")
|
|
print(f"\nSaved audit to {output.resolve()}")
|
|
return 1 if any(status != "MATCH" for status in result["status"]) else 0
|
|
|
|
|
|
def _import_set_actions(args: argparse.Namespace) -> int:
|
|
symbols = sorted({symbol.strip().upper().removesuffix(".BK") for symbol in args.symbols if symbol.strip()})
|
|
if not symbols:
|
|
raise ValueError("select at least one ticker to import")
|
|
aliases_path = Path(args.symbol_aliases) if args.symbol_aliases else None
|
|
aliases = pd.DataFrame(columns=["source_symbol", "price_symbol", "start_date", "end_date", "reason"])
|
|
if aliases_path and aliases_path.exists():
|
|
aliases = pd.read_csv(aliases_path, dtype=str).fillna("")
|
|
required = {"source_symbol", "price_symbol", "start_date", "end_date", "reason"}
|
|
if not required.issubset(aliases.columns):
|
|
raise ValueError(f"symbol alias CSV must contain {', '.join(sorted(required))}")
|
|
for column in ("source_symbol", "price_symbol"):
|
|
aliases[column] = aliases[column].str.upper().str.strip()
|
|
aliases["start_date"] = pd.to_datetime(aliases["start_date"], errors="coerce").dt.normalize()
|
|
aliases["end_date"] = pd.to_datetime(aliases["end_date"], errors="coerce").dt.normalize()
|
|
if aliases[["start_date", "end_date"]].isna().any().any() or (aliases["end_date"] <= aliases["start_date"]).any():
|
|
raise ValueError("symbol alias CSV contains invalid effective date intervals")
|
|
fetch_symbols = sorted(set(symbols) | set(aliases["source_symbol"].tolist()))
|
|
reference, issues = fetch_set_cash_dividend_reference(
|
|
fetch_symbols, start=args.start, end=args.end, delay_seconds=args.delay_seconds,
|
|
)
|
|
start_date, end_date = pd.Timestamp(args.start).normalize(), pd.Timestamp(args.end).normalize()
|
|
if not reference.empty:
|
|
reference["source_symbol"] = reference["symbol"].astype(str).str.upper()
|
|
reference["date"] = pd.to_datetime(reference["date"]).dt.normalize()
|
|
for alias in aliases.itertuples(index=False):
|
|
mask = ((reference["source_symbol"] == alias.source_symbol)
|
|
& (reference["date"] >= alias.start_date)
|
|
& (reference["date"] < alias.end_date))
|
|
reference.loc[mask, "symbol"] = alias.price_symbol
|
|
key = ["symbol", "action_type", "date"]
|
|
collapsed: list[dict[str, object]] = []
|
|
for _, group in reference.groupby(key, sort=False):
|
|
if group["value"].max() - group["value"].min() > 1e-9:
|
|
names = ", ".join(sorted(group["source_symbol"].astype(str).unique()))
|
|
raise SiamChartActionError(
|
|
f"aliased SET symbols disagree for {group.iloc[0]['symbol']} "
|
|
f"{pd.Timestamp(group.iloc[0]['date']).date()}: {names}"
|
|
)
|
|
row = group.iloc[0].to_dict()
|
|
row["source_symbol"] = ",".join(sorted(group["source_symbol"].astype(str).unique()))
|
|
row["source_url"] = " | ".join(sorted(group["source_url"].astype(str).unique()))
|
|
collapsed.append(row)
|
|
reference = pd.DataFrame(collapsed).sort_values(["symbol", "date"]).reset_index(drop=True)
|
|
else:
|
|
reference["source_symbol"] = pd.Series(dtype=str)
|
|
reference["coverage_start"] = start_date.date().isoformat()
|
|
reference["coverage_end"] = end_date.date().isoformat()
|
|
reference["coverage_complete"] = not issues
|
|
output = Path(args.output)
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
reference.to_csv(output, index=False)
|
|
manifest = {
|
|
"schema_version": 1,
|
|
"provider": "SET public Rights & Benefits history",
|
|
"source_page": "https://www.set.or.th/en/market/product/stock/quote/{symbol}/rights-benefits",
|
|
"source_endpoint": "https://www.set.or.th/api/set/stock/{symbol}/corporate-action?lang=en",
|
|
"start_inclusive": start_date.date().isoformat(),
|
|
"end_exclusive": end_date.date().isoformat(),
|
|
"requested_price_symbols": symbols,
|
|
"fetched_set_symbols": fetch_symbols,
|
|
"symbol_aliases": [
|
|
{"source_symbol": row.source_symbol, "price_symbol": row.price_symbol,
|
|
"start_date": row.start_date.date().isoformat(), "end_date": row.end_date.date().isoformat(),
|
|
"reason": row.reason}
|
|
for row in aliases.itertuples(index=False)
|
|
],
|
|
"coverage_complete": not issues,
|
|
"issues": issues,
|
|
}
|
|
manifest_path = Path(str(output) + ".manifest.json")
|
|
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(f"Imported {len(reference)} SET cash-dividend event(s) across {len(symbols)} price symbols "
|
|
f"({len(fetch_symbols)} SET ticker pages fetched).")
|
|
print(f"Saved SET reference to {output.resolve()}")
|
|
print(f"Saved coverage manifest to {manifest_path.resolve()}")
|
|
if issues:
|
|
print(f"\nSET symbols without reference data: {len(issues)}", file=sys.stderr)
|
|
for issue in issues:
|
|
print(f"- {issue}", file=sys.stderr)
|
|
return 1 if issues else 0
|
|
|
|
|
|
def _market_report(market: pd.DataFrame) -> None:
|
|
row = market.iloc[-1]
|
|
mco_z = row.get("mco_z", float("nan"))
|
|
sigma = f"{mco_z:.2f} sigma" if pd.notna(mco_z) else "warming up"
|
|
print("THAI MARKET")
|
|
print(f"Date {market.index[-1].date()}")
|
|
print(f"SET 21DMA {row.get('trend_state', 'UNKNOWN')}")
|
|
print(f"SET100EW 21DMA {row.get('ew_trend_state', 'UNKNOWN')}")
|
|
print(f"MCO {row.get('mco', float('nan')):.2f} ({sigma})" if pd.notna(row.get("mco", float("nan"))) else f"MCO {sigma}")
|
|
print(f"MCSI {row.get('mcsi_state', 'UNKNOWN')}")
|
|
print(f"Breadth A {int(row.get('advancing', 0))} / D {int(row.get('declining', 0))} / U {int(row.get('unchanged', 0))}")
|
|
print(f"\nREGIME {row.get('regime', 'UNKNOWN')}")
|
|
print(f"NEW RISK {'ALLOWED' if bool(row.get('new_risk_allowed', False)) else 'BLOCKED'}")
|
|
for reason in row.get("regime_reasons", []):
|
|
print(f" - {reason}")
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = _parser().parse_args(argv)
|
|
try:
|
|
if args.command == "download":
|
|
return _download(args)
|
|
if args.command == "import-siamchart":
|
|
return _import_siamchart(args)
|
|
if args.command == "import-siamchart-actions":
|
|
return _import_siamchart_actions(args)
|
|
if args.command == "verify-siamchart-actions":
|
|
return _verify_siamchart_actions(args)
|
|
if args.command == "import-set-actions":
|
|
return _import_set_actions(args)
|
|
if args.command == "webui":
|
|
from .webui import run_server
|
|
run_server(host=args.host, port=args.port, archives=args.archives, report_dir=args.report_dir,
|
|
start=args.start, initial_equity=args.initial_equity)
|
|
return 0
|
|
if args.command == "smoke":
|
|
config = AppConfig()
|
|
stock_bars, set_bars = synthetic_bars()
|
|
market, _, _ = build_market_context(stock_bars, set_bars, current_members=SMOKE_SYMBOLS,
|
|
research=config.research)
|
|
panel = build_feature_panel(stock_bars, set_bars, market, rs_weights=config.rs_weights,
|
|
volume_ratio=config.strategy.volume_pullback_ratio,
|
|
research=config.research)
|
|
scan = scan_candidates(panel, market, config=config, top_n=5)
|
|
result = run_backtest(panel, market, strategy=config.strategy,
|
|
rules=config.market_rules, costs=config.costs)
|
|
print("SYNTHETIC OFFLINE SMOKE - NOT MARKET PERFORMANCE")
|
|
print(f"bars={len(set_bars)} | price series={len(panel)} | SET index=yes | market context=yes")
|
|
print(f"indicators=ready | breadth/MCO/MCSI=ready | regime={market.iloc[-1]['regime']}")
|
|
print(f"scanner candidates={len(scan.candidates)} | backtest trades={len(result.trades)}")
|
|
print(f"total return={result.metrics['total_return']:.2%} | trim@2R={result.metrics['trim_2r_contribution']:,.0f} THB | runner={result.metrics['runner_contribution']:,.0f} THB")
|
|
return 0
|
|
if args.command == "model-lab":
|
|
if pd.Timestamp(args.end) <= pd.Timestamp(args.start):
|
|
raise ValueError("--end must be later than --start")
|
|
summary = run_model_lab(
|
|
args.archives, start=args.start, end=args.end, output_dir=args.output_dir,
|
|
action_dir=args.action_dir, set_index_csv=args.set_index_csv,
|
|
set_reference=args.set_reference, initial_equity=args.initial_equity,
|
|
)
|
|
print("MODEL SCREEN (fractional-share pass)")
|
|
for row in summary.itertuples(index=False):
|
|
print(f"{row.model:34s} {row.scenario:12s} "
|
|
f"return={row.total_return:7.1%} CAGR={row.cagr:6.1%} "
|
|
f"maxDD={row.max_drawdown:7.1%} Sharpe={row.sharpe_daily:5.2f}")
|
|
lot_report = Path(args.output_dir) / "board_lot_validation.csv"
|
|
if lot_report.exists():
|
|
lots = pd.read_csv(lot_report)
|
|
print("\nTOP CANDIDATES (100-share buy lots)")
|
|
for row in lots.itertuples(index=False):
|
|
print(f"{row.model:34s} {row.scenario:12s} "
|
|
f"return={row.total_return:7.1%} CAGR={row.cagr:6.1%} "
|
|
f"maxDD={row.max_drawdown:7.1%}")
|
|
primary_union = lots[lots["model"] == "primary_union_donchian_ema"]
|
|
if not primary_union.empty:
|
|
print("\nPRIMARY MODEL UNION (either Donchian or EMA signal)")
|
|
for row in primary_union.itertuples(index=False):
|
|
print(f"{row.scenario:12s} return={row.total_return:7.1%} "
|
|
f"CAGR={row.cagr:6.1%} maxDD={row.max_drawdown:7.1%}")
|
|
print(f"\nSaved model-lab reports to {Path(args.output_dir).resolve()}")
|
|
return 0
|
|
config = _load_config(args.config)
|
|
panel, market, set_bars, membership, _ = _prepare(args, config)
|
|
if args.command == "market":
|
|
_market_report(market)
|
|
return 0
|
|
out_dir = Path(args.output_dir)
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
if args.command == "scan":
|
|
when = pd.Timestamp(args.end) - pd.Timedelta(days=1) if args.end else None
|
|
if when is not None:
|
|
panel = {symbol: frame.loc[frame.index <= when] for symbol, frame in panel.items()}
|
|
market = market.loc[market.index <= when]
|
|
result = scan_candidates(panel, market, config=config, top_n=args.top,
|
|
sectors={symbol: membership.sector(symbol, market.index[-1])
|
|
for symbol in panel if membership} if membership else None)
|
|
_market_report(market)
|
|
print("\nFOCUS LIST")
|
|
if result.candidates.empty:
|
|
print("No liquid candidates meet the configured research filters.")
|
|
else:
|
|
view = result.candidates[["ticker", "score", "rs", "atr_distance", "setup", "entry", "stop", "shares"]]
|
|
print(view.to_string(index=False, float_format=lambda value: f"{value:.2f}"))
|
|
print("\nDETAILS")
|
|
for row in result.candidates.head(3).to_dict("records"):
|
|
print(f"\n{row['ticker']} Score {row['score']:.1f} RS {row['rs']:.1f}")
|
|
print(f"Setup {row['setup']} | Entry ref {row['entry']:.2f} | Stop {row['stop']:.2f} | R {row['initial_R']}")
|
|
print(f"Risk {row['risk_pct']:.2%} | Shares {row['shares']} | Position {row['position_value_thb']:,.0f} THB")
|
|
print("Signals: " + (", ".join(row["reasons"]) or "Watch only"))
|
|
if args.json_out:
|
|
result.write_json(args.json_out)
|
|
print(f"\nQwen-ready JSON: {args.json_out}")
|
|
return 0
|
|
if args.command == "strategy-overlays":
|
|
if not args.start or not args.end:
|
|
raise ValueError("strategy-overlays requires explicit --start and --end dates")
|
|
window_start, window_end = pd.Timestamp(args.start), pd.Timestamp(args.end)
|
|
if window_end <= window_start:
|
|
raise ValueError("--end must be later than --start")
|
|
periods: list[tuple[str, str, str]] = [("full", args.start, args.end)]
|
|
for year in range(window_start.year, window_end.year + 1):
|
|
year_start = max(window_start, pd.Timestamp(year=year, month=1, day=1))
|
|
year_end = min(window_end, pd.Timestamp(year=year + 1, month=1, day=1))
|
|
if year_start < year_end:
|
|
periods.append((str(year), year_start.date().isoformat(), year_end.date().isoformat()))
|
|
variants: list[tuple[str, dict[str, object]]] = [
|
|
("baseline", {}),
|
|
("weak_retest_no_1r_after_5_sessions", {"stalled_retest_sessions": 5}),
|
|
("weak_retest_no_1r_after_8_sessions", {"stalled_retest_sessions": 8}),
|
|
("weak_retest_no_1r_after_10_sessions", {"stalled_retest_sessions": 10}),
|
|
("runner_floor_0r_after_2r_trim", {"runner_protection_r": 0.0}),
|
|
("weak_retest_5_sessions_plus_runner_floor_0r",
|
|
{"stalled_retest_sessions": 5, "runner_protection_r": 0.0}),
|
|
]
|
|
summary_rows: list[dict[str, object]] = []
|
|
for variant, overlay in variants:
|
|
for period, period_start, period_end in periods:
|
|
result = run_backtest(
|
|
panel, market, strategy=config.strategy, rules=config.market_rules,
|
|
costs=config.costs, start=period_start, end=period_end, **overlay,
|
|
)
|
|
run_dir = out_dir / "strategy_overlays" / variant / period
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
metrics_json = {
|
|
key: (value if math.isfinite(value) else None)
|
|
for key, value in result.metrics.items()
|
|
}
|
|
(run_dir / "backtest_metrics.json").write_text(
|
|
json.dumps(metrics_json, indent=2, allow_nan=False), encoding="utf-8"
|
|
)
|
|
result.trades.to_csv(run_dir / "trades.csv", index=False)
|
|
result.equity_curve.to_csv(run_dir / "equity_curve.csv", index_label="date")
|
|
result.fills.to_csv(run_dir / "fills.csv", index=False)
|
|
pd.DataFrame(result.skipped, columns=["symbol", "date", "reason"]).to_csv(
|
|
run_dir / "skipped_entries.csv", index=False
|
|
)
|
|
summary_rows.append({
|
|
"variant": variant, "period": period, "start": period_start,
|
|
"end_exclusive": period_end, "trades": len(result.trades),
|
|
**result.metrics,
|
|
})
|
|
summary = pd.DataFrame(summary_rows)
|
|
summary_path = out_dir / "strategy_overlays" / "summary.csv"
|
|
summary.to_csv(summary_path, index=False)
|
|
readme = (
|
|
"# Strategy exit overlays\n\n"
|
|
f"Signal window: {window_start.date()} through {(window_end - pd.Timedelta(days=1)).date()} inclusive.\n\n"
|
|
"The baseline and overlays use the same PrimeThai entries, position sizing, execution costs, "
|
|
"and corporate-action inputs. The stalled-retest overlays exit a WEAKNESS/21DMA Retest at "
|
|
"the next open after 5, 8, or 10 sessions if its intraday maximum has not reached +1R. "
|
|
"The runner-floor overlay activates a stop at entry (0R) only from the session after a 2R trim. "
|
|
"The combined overlay applies both rules. Defaults remain unchanged.\n\n"
|
|
"Each calendar slice starts with fresh equity and liquidates at its own end; slices are not "
|
|
"additive. This short, repeatedly examined window is exploratory and is not an untouched "
|
|
"out-of-sample validation. Results do not establish a durable edge or predict live returns.\n"
|
|
)
|
|
(out_dir / "strategy_overlays" / "README.md").write_text(readme, encoding="utf-8")
|
|
view = summary[["variant", "period", "trades", "total_return", "max_drawdown",
|
|
"profit_factor", "expectancy_per_trade", "average_holding_period"]]
|
|
print("PRIMETHAI EXIT OVERLAYS - SAME ENTRIES, NEXT-OPEN EXECUTION")
|
|
print(view.to_string(index=False, float_format=lambda value: f"{value:.4f}"))
|
|
print(f"\nSaved detailed runs to {(out_dir / 'strategy_overlays').resolve()}")
|
|
return 0
|
|
if args.command == "ema-crossover":
|
|
pairs = [_parse_ema_pair(value) for value in args.ema_pairs]
|
|
if len(set(pairs)) != len(pairs):
|
|
raise ValueError("--ema-pairs contains a duplicate pair")
|
|
if not args.market_filters or len(set(args.market_filters)) != len(args.market_filters):
|
|
raise ValueError("--market-filters must be non-empty and contain no duplicates")
|
|
if not args.exit_modes or len(set(args.exit_modes)) != len(args.exit_modes):
|
|
raise ValueError("--exit-modes must be non-empty and contain no duplicates")
|
|
if (not args.max_loss_pcts
|
|
or any(not math.isfinite(value) or not 0 <= value < 1 for value in args.max_loss_pcts)
|
|
or len(set(args.max_loss_pcts)) != len(args.max_loss_pcts)):
|
|
raise ValueError("--max-loss-pcts must be unique values from 0 (off) to less than 1")
|
|
periods = _ema_periods(args, market)
|
|
position_pct = (config.strategy.max_position_pct if args.max_position_pct is None
|
|
else args.max_position_pct)
|
|
min_price = config.strategy.min_price if args.min_price is None else args.min_price
|
|
min_turnover = (config.strategy.min_value_turnover_thb
|
|
if args.min_value_turnover_thb is None else args.min_value_turnover_thb)
|
|
summary_rows: list[dict[str, object]] = []
|
|
for fast, slow in pairs:
|
|
for market_filter in args.market_filters:
|
|
for exit_mode in args.exit_modes:
|
|
for stop_pct in args.max_loss_pcts:
|
|
stop_setting = None if stop_pct == 0 else stop_pct
|
|
for label, period_start, period_end in periods:
|
|
result = run_ema_crossover_backtest(
|
|
panel, market, fast=fast, slow=slow, strategy=config.strategy,
|
|
rules=config.market_rules, costs=config.costs,
|
|
start=period_start, end=period_end,
|
|
max_position_pct=position_pct, min_price=min_price,
|
|
min_value_turnover_thb=min_turnover,
|
|
market_filter=market_filter, exit_mode=exit_mode,
|
|
max_loss_pct=stop_setting,
|
|
)
|
|
stop_label = "none" if stop_setting is None else f"{stop_setting:.0%}"
|
|
variant = f"filter_{market_filter}__exit_{exit_mode}__stop_{stop_label}"
|
|
run_dir = out_dir / f"ema_{fast}_{slow}" / variant / label
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
metrics_json = {
|
|
key: (value if math.isfinite(value) else None)
|
|
for key, value in result.metrics.items()
|
|
}
|
|
(run_dir / "backtest_metrics.json").write_text(
|
|
json.dumps(metrics_json, indent=2, allow_nan=False), encoding="utf-8"
|
|
)
|
|
result.trades.to_csv(run_dir / "trades.csv", index=False)
|
|
result.equity_curve.to_csv(run_dir / "equity_curve.csv", index_label="date")
|
|
result.fills.to_csv(run_dir / "fills.csv", index=False)
|
|
pd.DataFrame(result.skipped, columns=["symbol", "date", "reason"]).to_csv(
|
|
run_dir / "skipped_entries.csv", index=False
|
|
)
|
|
summary_rows.append({
|
|
"ema_pair": f"{fast}:{slow}", "market_filter": market_filter,
|
|
"exit_mode": exit_mode, "max_loss_pct": stop_pct,
|
|
"period": label, "start": period_start,
|
|
"end_exclusive": period_end, "trades": len(result.trades),
|
|
**result.metrics,
|
|
})
|
|
summary = pd.DataFrame(summary_rows)
|
|
summary.to_csv(out_dir / "summary.csv", index=False)
|
|
print("EMA CROSSOVER BACKTEST - CLOSE SIGNAL, NEXT-OPEN FILL")
|
|
print(summary[["ema_pair", "market_filter", "exit_mode", "max_loss_pct", "period",
|
|
"trades", "total_return", "max_drawdown",
|
|
"profit_factor", "win_rate", "exposure_pct"]].to_string(
|
|
index=False, float_format=lambda value: f"{value:.4f}"
|
|
))
|
|
print(f"\nSaved detailed runs and summary to {out_dir.resolve()}")
|
|
print(f"Assumptions: at most {position_pct:.1%} of equity per stock; simultaneous entries share available cash equally; "
|
|
f"{min_price:g} THB minimum price; {min_turnover:,.0f} THB minimum 20-day average turnover; "
|
|
f"market filters={', '.join(args.market_filters)}; exits={', '.join(args.exit_modes)}; "
|
|
f"hard stops={', '.join('off' if value == 0 else f'{value:.0%}' for value in args.max_loss_pcts)}.")
|
|
return 0
|
|
backtest = run_backtest(panel, market, strategy=config.strategy, rules=config.market_rules,
|
|
costs=config.costs, start=args.start, end=args.end)
|
|
if args.command == "backtest":
|
|
(out_dir / "backtest_metrics.json").write_text(json.dumps(backtest.metrics, indent=2), encoding="utf-8")
|
|
backtest.trades.to_csv(out_dir / "trades.csv", index=False)
|
|
backtest.equity_curve.to_csv(out_dir / "equity_curve.csv", index_label="date")
|
|
backtest.fills.to_csv(out_dir / "fills.csv", index=False)
|
|
pd.DataFrame(backtest.skipped, columns=["symbol", "date", "reason"]).to_csv(
|
|
out_dir / "skipped_entries.csv", index=False
|
|
)
|
|
print(json.dumps(backtest.metrics, indent=2))
|
|
print(f"\nTrades: {len(backtest.trades)} | trim @ 2R: {backtest.metrics['trim_2r_contribution']:,.2f} THB | runner: {backtest.metrics['runner_contribution']:,.2f} THB")
|
|
print(f"Saved reports to {out_dir.resolve()}")
|
|
return 0
|
|
if args.command == "experiment":
|
|
comparison, _ = run_ablations(panel, market, base=config.strategy, rules=config.market_rules, costs=config.costs)
|
|
holdouts, _ = run_chronological_holdouts(panel, market, strategy=config.strategy, rules=config.market_rules, costs=config.costs)
|
|
walk = rolling_walk_forward(panel, market, strategy=config.strategy, rules=config.market_rules, costs=config.costs)
|
|
comparison.to_csv(out_dir / "ablation.csv", index=False)
|
|
holdouts.to_csv(out_dir / "holdouts.csv", index=False)
|
|
walk.to_csv(out_dir / "walk_forward.csv", index=False)
|
|
print("ABLATION")
|
|
print(comparison[["experiment", "total_return", "max_drawdown", "win_rate", "expectancy_r", "profit_factor", "trades"]].to_string(index=False))
|
|
print("\nCHRONOLOGICAL HOLDOUTS")
|
|
print(holdouts[["period", "start", "end", "total_return", "max_drawdown", "trades"]].to_string(index=False))
|
|
print(f"\nSaved tables to {out_dir.resolve()}")
|
|
return 0
|
|
except (ValueError, MembershipError, SymbolDataError, SiamChartActionError, OSError, KeyError) as exc:
|
|
print(f"Error: {exc}", file=sys.stderr)
|
|
return 2
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
with warnings.catch_warnings(record=False):
|
|
warnings.simplefilter("default")
|
|
raise SystemExit(main())
|