from __future__ import annotations import importlib.util import sys from pathlib import Path import pandas as pd ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from okx_codex_trader.models import Candle from scripts import search_eth_price_twap_variants as search from scripts.explore_ultrashort import ( CANDLE_CACHE_DIR, INITIAL_EQUITY, LEVERAGE, build_rsi2_long_guarded_price_twap_candidate, cost_adjusted_trade_equity_frame, history_bars_for_years, load_cached_candles, recent_horizon_metrics_from_equity, ) SYMBOL = "ETH-USDT-SWAP" BAR = "15m" MAKER_TAKER_COST = 0.0021 OUTPUT_DIR = Path("reports/eth-exploration") CSV_PATH = OUTPUT_DIR / "eth-twap-evaluation-diff.csv" MD_PATH = OUTPUT_DIR / "eth-twap-evaluation-diff.md" HORIZONS = ( ("3y", pd.DateOffset(years=3)), ("1y", pd.DateOffset(years=1)), ("6m", pd.DateOffset(months=6)), ("3m", pd.DateOffset(months=3)), ) DEEP_SPEC = { "trend_sma": 50, "rsi_threshold": 3.0, "exit_rsi": 55.0, "stop_loss_pct": 0.008, "max_hold_bars": 48, "entry_offsets": (0.002, 0.005, 0.008), "entry_valid_bars": 3, "fill_buffer": 0.0, } def load_report_module(): path = Path(__file__).resolve().with_name("generate_ultrashort_report.py") spec = importlib.util.spec_from_file_location("generate_ultrashort_report", path) if spec is None or spec.loader is None: raise RuntimeError("cannot load generate_ultrashort_report.py") module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module def cached_history(years: float) -> list[Candle]: candles, _ = load_cached_candles(CANDLE_CACHE_DIR, SYMBOL, BAR) if not candles: raise RuntimeError(f"missing cached candles: {SYMBOL} {BAR}") requested_bars = history_bars_for_years(BAR, years) return candles[-requested_bars:] if len(candles) > requested_bars else candles def equity_frame_from_search_result(result) -> pd.DataFrame: return search.equity_frame(result) def metric_rows(label: str, candles: list[Candle], trade_count: int, gross_equity: pd.DataFrame, net_equity: pd.DataFrame) -> list[dict[str, object]]: gross_return = float(gross_equity["equity"].iloc[-1] / gross_equity["equity"].iloc[0] - 1.0) net_return = float(net_equity["equity"].iloc[-1] / net_equity["equity"].iloc[0] - 1.0) rows: list[dict[str, object]] = [ { "evaluation": label, "window": "total", "first_candle": search._format_ts(candles[0].ts), "last_candle": search._format_ts(candles[-1].ts), "bars": len(candles), "trades": trade_count, "gross_return": gross_return, "maker_taker_net_return": net_return, } ] horizons = recent_horizon_metrics_from_equity(net_equity, candles[-1].ts, HORIZONS) for row in horizons.to_dict("records"): rows.append( { "evaluation": label, "window": row["horizon"], "first_candle": row["horizon_start"], "last_candle": row["horizon_end"], "bars": "", "trades": trade_count, "gross_return": "", "maker_taker_net_return": float(row["net_total_return"]), } ) return rows def pct(value: object) -> str: if value == "": return "" return f"{float(value) * 100:.2f}%" def markdown_table(frame: pd.DataFrame) -> str: display = frame.copy() display["gross_return"] = display["gross_return"].map(pct) display["maker_taker_net_return"] = display["maker_taker_net_return"].map(pct) columns = [ "evaluation", "window", "first_candle", "last_candle", "bars", "trades", "gross_return", "maker_taker_net_return", ] rows = [["" if pd.isna(value) else str(value) for value in row] for row in display[columns].itertuples(index=False, name=None)] return "\n".join( [ "| " + " | ".join(columns) + " |", "| " + " | ".join("---" for _ in columns) + " |", *["| " + " | ".join(row) + " |" for row in rows], ] ) def main() -> int: report = load_report_module() report_candles = report.load_cached_history(report.load_explore_module(), SYMBOL, BAR, history_bars_for_years(BAR, 10.0)) report_candidate = build_rsi2_long_guarded_price_twap_candidate( int(DEEP_SPEC["trend_sma"]), float(DEEP_SPEC["rsi_threshold"]), float(DEEP_SPEC["exit_rsi"]), float(DEEP_SPEC["stop_loss_pct"]), int(DEEP_SPEC["max_hold_bars"]), tuple(DEEP_SPEC["entry_offsets"]), int(DEEP_SPEC["entry_valid_bars"]), float(DEEP_SPEC["fill_buffer"]), ) report_gross = report_candidate.run( candles=report_candles, leverage=LEVERAGE, warmup_bars=report_candidate.warmup_bars, ) report_gross_equity = cost_adjusted_trade_equity_frame(report_gross, 0.0) report_net = cost_adjusted_trade_equity_frame(report_gross, MAKER_TAKER_COST) search_candles = cached_history(search.YEARS) search_gross = search.run_price_twap_segment( candles=search_candles, spec=DEEP_SPEC, roundtrip_cost_on_margin=0.0, ) search_gross_equity = equity_frame_from_search_result(search_gross) search_net_result = search.run_price_twap_segment( candles=search_candles, spec=DEEP_SPEC, roundtrip_cost_on_margin=MAKER_TAKER_COST, ) search_net = equity_frame_from_search_result(search_net_result) rows = [ *metric_rows("main_report_10y_eval", report_candles, report_gross.trade_count, report_gross_equity, report_net), *metric_rows("search_script_3y_eval", search_candles, search_gross.trade_count, search_gross_equity, search_net), ] frame = pd.DataFrame(rows) OUTPUT_DIR.mkdir(parents=True, exist_ok=True) frame.to_csv(CSV_PATH, index=False) MD_PATH.write_text( "\n".join( [ "# ETH RSI2 Price TWAP deep evaluation diff", "", "Parameter under test: `trend_sma=50, rsi_threshold=3.0, exit_rsi=55.0, stop_loss_pct=0.008, max_hold_bars=48, entry_offsets=(0.002, 0.005, 0.008), entry_valid_bars=3, fill_buffer=0.0`.", "", markdown_table(frame), "", "## Readout", "", "- `main_report_10y_eval` matches `scripts/generate_ultrashort_report.py`: load up to 10 years, run the shared `explore_ultrashort` candidate gross, then apply maker_taker cost with `cost_weight` per closed trade.", "- `search_script_3y_eval` matches `scripts/search_eth_price_twap_variants.py`: load 3 years, run its local price-TWAP segment, and deduct maker_taker cost inside each close.", "- The prior search grid did not include the main-report ETH deep parameter. It searched `trend_sma in {80,160}`, `rsi_threshold in {5,8,10}`, `entry_valid_bars in {2,4}`, and offset sets `(0.001,0.003,0.005)` / `(0.003,0.006,0.009)`, not `trend_sma=50`, `rsi_threshold=3`, deep offsets `(0.002,0.005,0.008)`, `entry_valid_bars=3`.", "- Therefore the sign mismatch is primarily parameter universe plus total-sample scope: the main report's cost table is 10-year total, while the search table was a different 3-year grid sorted by maker_taker `net_calmar` then `net_annualized_return`.", "", "Adopt the main-report evaluation口径 for reporting selected strategies: shared `explore_ultrashort` runner, 10-year cached history for total cost sensitivity, and explicit recent horizons for 3y/1y/6m/3m.", "", ] ), encoding="utf-8", ) print(CSV_PATH) print(MD_PATH) print(frame.to_string(index=False)) return 0 if __name__ == "__main__": raise SystemExit(main())