NexQuant/scripts/nexquant_live_priceaction.py

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