diff --git a/scripts/nexquant_live_priceaction.py b/scripts/nexquant_live_priceaction.py new file mode 100644 index 00000000..9f8b36ed --- /dev/null +++ b/scripts/nexquant_live_priceaction.py @@ -0,0 +1,185 @@ +#!/usr/bin/env python3 +"""Live Price-Action Strategy Pipeline — No LLM, No Factors. + +Generates daily signals from Donchian + MACD portfolio, executes via risk +backtest, and optionally sends signals to live trading. + +Usage: + python scripts/nexquant_live_priceaction.py # Generate today's signal + python scripts/nexquant_live_priceaction.py --daemon # Run continuously + python scripts/nexquant_live_priceaction.py --backfill # Full historical backtest +""" + +import json +import os +import sys +import time +from datetime import datetime, timedelta +from pathlib import Path + +import numpy as np +import pandas as pd + +PROJECT = Path(__file__).resolve().parent.parent +OHLCV_PATH = Path(os.getenv("PREDIX_OHLCV_PATH", + str(PROJECT / "git_ignore_folder" / "intraday_pv_all.h5"))) +SIGNAL_PATH = PROJECT / "git_ignore_folder" / "priceaction_signal.json" +RESULTS_DIR = PROJECT / "results" / "reports" + +# Portfolio config +STRATEGIES = [ + {"name": "Donchian(30,1)", "type": "donchian", "period": 30, "hold": 1}, + {"name": "MACD(3,15,3)", "type": "macd", "fast": 3, "slow": 15, "signal_period": 3}, +] + +VOTE_THRESHOLD = 0.25 + + +def load_close() -> tuple[pd.Series, pd.Series]: + """Load 1-min and daily close prices.""" + df = pd.read_hdf(OHLCV_PATH, key="data") + close = df.xs("EURUSD", level="instrument")["$close"].sort_index() + daily = close.resample("D").last().dropna() + return close, daily + + +def donchian_signal(daily: pd.Series, period: int, hold: int) -> pd.Series: + """Donchian channel breakout signal (daily).""" + high = daily.rolling(period).max() + low = daily.rolling(period).min() + s = pd.Series(0, index=daily.index) + s[daily > high.shift(1)] = 1 + s[daily < low.shift(1)] = -1 + return s.replace(0, np.nan).ffill(limit=hold).fillna(0).astype(int).clip(-1, 1) + + +def macd_signal(daily: pd.Series, fast: int, slow: int, signal_period: int) -> pd.Series: + """MACD crossover signal (daily).""" + ema_fast = daily.ewm(span=fast, adjust=False).mean() + ema_slow = daily.ewm(span=slow, adjust=False).mean() + macd_line = ema_fast - ema_slow + sig_line = macd_line.ewm(span=signal_period, adjust=False).mean() + s = pd.Series(0, index=daily.index) + s[macd_line > sig_line] = 1 + s[macd_line < sig_line] = -1 + return s.fillna(0).astype(int).clip(-1, 1) + + +def compute_portfolio_signal(daily: pd.Series) -> pd.Series: + """Compute majority-vote portfolio signal.""" + signals = [] + for cfg in STRATEGIES: + if cfg["type"] == "donchian": + sig = donchian_signal(daily, cfg["period"], cfg["hold"]) + elif cfg["type"] == "macd": + sig = macd_signal(daily, cfg["fast"], cfg["slow"], cfg["signal_period"]) + else: + continue + signals.append(sig) + + if not signals: + return pd.Series(0, index=daily.index) + + port = pd.DataFrame({f"s{i}": s for i, s in enumerate(signals)}).dropna() + vote = port.mean(axis=1) + result = pd.Series(0, index=vote.index) + result[vote > VOTE_THRESHOLD] = 1 + result[vote < -VOTE_THRESHOLD] = -1 + result.name = "signal" + return result + + +def get_todays_signal() -> dict: + """Generate today's trading signal.""" + close, daily = load_close() + portfolio_signal = compute_portfolio_signal(daily) + + # Latest signal + latest = portfolio_signal.iloc[-1] + direction = {1: "LONG", -1: "SHORT", 0: "NEUTRAL"}[int(latest)] + + # Last signal change + changes = portfolio_signal.diff().abs() + last_change_idx = changes[changes > 0].index[-1] if (changes > 0).any() else None + days_in_position = (daily.index[-1] - last_change_idx).days if last_change_idx is not None else 0 + + result = { + "timestamp": datetime.now().isoformat(), + "date": str(daily.index[-1].date()), + "signal": int(latest), + "direction": direction, + "days_in_position": days_in_position, + "strategies": {cfg["name"]: int( + donchian_signal(daily, cfg["period"], cfg["hold"]).iloc[-1] if cfg["type"] == "donchian" + else macd_signal(daily, cfg["fast"], cfg["slow"], cfg["signal_period"]).iloc[-1] + ) for cfg in STRATEGIES}, + } + + SIGNAL_PATH.parent.mkdir(parents=True, exist_ok=True) + SIGNAL_PATH.write_text(json.dumps(result, indent=2)) + + return result + + +def run_backfill(): + """Run full historical backtest and save report.""" + print("Running full historical backtest...") + close, daily = load_close() + signal = compute_portfolio_signal(daily) + + # ffill to 1-min + sig_1min = signal.reindex(close.index).ffill().fillna(0).astype(int).clip(-1, 1) + + from rdagent.components.backtesting.vbt_backtest import backtest_signal, backtest_signal_risk + + bt = backtest_signal(close=close, signal=sig_1min) + bt_risk = backtest_signal_risk(close=close, signal=sig_1min, risk_pct=0.0035, oos_start=None, wf_rolling=True) + + report = { + "strategy": "Donchian(30,1) + MACD(3,15,3) Majority-Vote", + "timestamp": datetime.now().isoformat(), + "backtest": { + "sharpe": round(bt["sharpe"], 2), + "monthly_return_pct": round(bt["monthly_return_pct"], 2), + "max_drawdown": round(bt["max_drawdown"], 4), + "n_trades": bt["n_trades"], + "win_rate": round(bt["win_rate"], 4), + }, + "risk_backtest": { + "sharpe": round(bt_risk.get("sharpe", 0), 2), + "monthly_pct": round(bt_risk.get("monthly_return_pct", 0), 2), + "max_dd": round(bt_risk.get("max_drawdown", 0), 4), + "wf_consistency": round(bt_risk.get("wf_oos_consistency", 0), 4), + }, + } + + RESULTS_DIR.mkdir(parents=True, exist_ok=True) + path = RESULTS_DIR / f"backfill_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" + path.write_text(json.dumps(report, indent=2)) + + print(f"\n{'='*50}") + print(f" Sharpe: {bt['sharpe']:.2f}") + print(f" Monthly: {bt['monthly_return_pct']:.2f}%") + print(f" Max DD: {bt['max_drawdown']:.4f}") + print(f" Trades: {bt['n_trades']}") + print(f" Win Rate: {bt['win_rate']:.1%}") + print(f" Report saved: {path}") + print(f"{'='*50}") + + +def main(): + if "--backfill" in sys.argv: + run_backfill() + elif "--daemon" in sys.argv: + print("Daemon mode — generating signals every 5 minutes...") + while True: + result = get_todays_signal() + print(f" [{result['timestamp']}] {result['direction']:>8s} ({result['days_in_position']}d in position)") + time.sleep(300) + else: + result = get_todays_signal() + print(json.dumps(result, indent=2)) + + +if __name__ == "__main__": + main()