Spaces:

Ypeng12
/

quant-ai

Running

quant-ai / backend /simulation /simulate_sndk.py

feat(ai): implement dedicated per-ticker simulation and regime classification engine

0581467 2 months ago

7.02 kB

# backend/simulation/simulate_sndk.py
"""
Dedicated Ticker Simulation: SNDK (High-Momentum Breakout & Trend-Holding Leader)
Zero-overlap independent simulation for SNDK.
Focus:
- Avoids 30s micro-panic exits.
- Adheres to GP Saggese's 5-30m+ wave holding principle.
- Leverages BULL_EXPANSION regime: buys momentum pullbacks near VWAP/EMA9, rides trend with trailing stop.
"""

import sys
import os
import yfinance as yf
import pandas as pd
import numpy as np

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))
from backend.app.ml.ticker_regime_engine import TickerRegimeEngine, MarketRegime

def run_sndk_simulation(period="5d", interval="5m", initial_capital=100000.0):
    print("=" * 80)
    print("      【SNDK 独立专用量化回测与模拟系统 (DEDICATED SIMULATION: SNDK)】")
    print("      核心特征: 高动量、强单边趋势、5~45分钟波段锁仓、拒绝秒级微观假摔洗盘")
    print(f"      初始本金: ${initial_capital:,.2f} | 数据周期: {period} (Interval: {interval})")
    print("=" * 80 + "\n")

    df = yf.download("SNDK", period=period, interval=interval, progress=False)
    if df.empty:
        print("Error: Could not retrieve SNDK data.")
        return

    # Clean multi-index columns if present
    if isinstance(df.columns, pd.MultiIndex):
        df.columns = [col[0] for col in df.columns]

    close_s = df["Close"].squeeze()
    open_s = df["Open"].squeeze()
    high_s = df["High"].squeeze()
    low_s = df["Low"].squeeze()
    vol_s = df["Volume"].squeeze()

    df["VWAP"] = TickerRegimeEngine.calculate_vwap(df)
    df["EMA_9"] = close_s.ewm(span=9, adjust=False).mean()
    df["EMA_21"] = close_s.ewm(span=21, adjust=False).mean()

    # Simulation State
    capital = initial_capital
    position_shares = 0
    entry_price = 0.0
    entry_time = None
    trades = []
    peak_price = 0.0

    # Iterate through bars
    for i in range(20, len(df)):
        sub_df = df.iloc[:i+1]
        regime_info = TickerRegimeEngine.analyze_ticker_regime(sub_df, "SNDK")

        current_time = df.index[i]
        bar_open = float(open_s.iloc[i])
        bar_high = float(high_s.iloc[i])
        bar_low = float(low_s.iloc[i])
        bar_close = float(close_s.iloc[i])
        bar_vwap = float(df["VWAP"].iloc[i])
        bar_ema9 = float(df["EMA_9"].iloc[i])
        bar_ema21 = float(df["EMA_21"].iloc[i])

        time_str = str(current_time)
        is_eod = "15:50" in time_str or "15:55" in time_str

        # 1. Check Exit if holding
        if position_shares > 0:
            peak_price = max(peak_price, bar_high)
            pnl_pct = (bar_close - entry_price) / entry_price * 100.0

            # Dynamic Trailing Exit for SNDK (GP Saggese wave rule: don't exit on micro dips)
            # Exit condition 1: Hard trend break (Close falls below EMA21 by > 0.6% AND below VWAP)
            # Exit condition 2: Large runner trailing stop (if up > 3.0%, trail 1.5% from peak)
            is_trend_broken = (bar_close < bar_ema21 * 0.994 and bar_close < bar_vwap)
            is_profit_trail_hit = (pnl_pct >= 2.5 and (peak_price - bar_close) / peak_price * 100.0 >= 1.5)

            if is_trend_broken or is_profit_trail_hit or is_eod:
                exit_price = bar_close
                trade_pnl = position_shares * (exit_price - entry_price)
                capital += position_shares * exit_price
                return_pct = (exit_price - entry_price) / entry_price * 100.0
                reason = "EOD Flatten" if is_eod else ("Trailing Stop Peak Lock" if is_profit_trail_hit else "Trend Life-Line Break")

                trades.append({
                    "entry_time": entry_time,
                    "exit_time": current_time,
                    "side": "LONG",
                    "shares": position_shares,
                    "entry_price": entry_price,
                    "exit_price": exit_price,
                    "pnl_$": trade_pnl,
                    "return_%": return_pct,
                    "reason": reason
                })
                position_shares = 0
                entry_price = 0.0

        # 2. Check Entry if flat
        elif position_shares == 0 and not is_eod:
            # SNDK Entry Rule: In BULL_EXPANSION or strong upward momentum pullback to EMA9/VWAP
            if regime_info.regime == MarketRegime.BULL_EXPANSION:
                # Enter when price is comfortably above VWAP and testing EMA9
                if bar_close >= bar_vwap and bar_close >= bar_ema9 * 0.998:
                    # Allocate 90% buying power into leader
                    alloc_capital = capital * 0.90
                    position_shares = int(alloc_capital / bar_close)
                    if position_shares > 0:
                        entry_price = bar_close
                        entry_time = current_time
                        peak_price = bar_close
                        capital -= position_shares * bar_close

    # Close open position at end
    if position_shares > 0:
        final_close = float(close_s.iloc[-1])
        trade_pnl = position_shares * (final_close - entry_price)
        capital += position_shares * final_close
        return_pct = (final_close - entry_price) / entry_price * 100.0
        trades.append({
            "entry_time": entry_time,
            "exit_time": df.index[-1],
            "side": "LONG",
            "shares": position_shares,
            "entry_price": entry_price,
            "exit_price": final_close,
            "pnl_$": trade_pnl,
            "return_%": return_pct,
            "reason": "Market Close Final Mark"
        })

    # Summary Statistics
    df_trades = pd.DataFrame(trades)
    print("=" * 80)
    print("   【SNDK 独立交易明细 (INDIVIDUAL TRADE LOG)】")
    print("=" * 80)
    if not df_trades.empty:
        for idx, row in df_trades.iterrows():
            win_icon = "🟢 盈利" if row["pnl_$"] > 0 else "🔴 亏损"
            print(f"  Trade #{idx+1} | {row['entry_time']} -> {row['exit_time']} | "
                  f"Entry: ${row['entry_price']:.2f} | Exit: ${row['exit_price']:.2f} | "
                  f"PnL: ${row['pnl_$']:+,.2f} ({row['return_%']:+.2f}%) | {win_icon} ({row['reason']})")
    else:
        print("  No trades triggered under strict regime criteria.")

    total_pnl = capital - initial_capital
    total_ret_pct = total_pnl / initial_capital * 100.0
    win_rate = (df_trades["pnl_$"] > 0).mean() * 100.0 if not df_trades.empty else 0.0

    print("\n" + "=" * 80)
    print("   【SNDK 独立回测指标总结】")
    print("=" * 80)
    print(f"  • 期末总资产 (Ending Equity):      ${capital:,.2f}")
    print(f"  • 累计总净利润 (Net Total PnL):    ${total_pnl:+,.2f}")
    print(f"  • 总收益率 (Total Net Return):     {total_ret_pct:+.2f}%")
    print(f"  • 胜率 (Win Rate):                 {win_rate:.1f}% ({len(df_trades)} 笔交易)")
    print("=" * 80 + "\n")

if __name__ == "__main__":
    run_sndk_simulation()

	# backend/simulation/simulate_sndk.py
	"""
	Dedicated Ticker Simulation: SNDK (High-Momentum Breakout & Trend-Holding Leader)
	Zero-overlap independent simulation for SNDK.
	Focus:
	- Avoids 30s micro-panic exits.
	- Adheres to GP Saggese's 5-30m+ wave holding principle.
	- Leverages BULL_EXPANSION regime: buys momentum pullbacks near VWAP/EMA9, rides trend with trailing stop.
	"""

	import sys
	import os
	import yfinance as yf
	import pandas as pd
	import numpy as np

	sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))
	from backend.app.ml.ticker_regime_engine import TickerRegimeEngine, MarketRegime

	def run_sndk_simulation(period="5d", interval="5m", initial_capital=100000.0):
	print("=" * 80)
	print(" 【SNDK 独立专用量化回测与模拟系统 (DEDICATED SIMULATION: SNDK)】")
	print(" 核心特征: 高动量、强单边趋势、5~45分钟波段锁仓、拒绝秒级微观假摔洗盘")
	print(f" 初始本金: ${initial_capital:,.2f} \| 数据周期: {period} (Interval: {interval})")
	print("=" * 80 + "\n")

	df = yf.download("SNDK", period=period, interval=interval, progress=False)
	if df.empty:
	print("Error: Could not retrieve SNDK data.")
	return

	# Clean multi-index columns if present
	if isinstance(df.columns, pd.MultiIndex):
	df.columns = [col[0] for col in df.columns]

	close_s = df["Close"].squeeze()
	open_s = df["Open"].squeeze()
	high_s = df["High"].squeeze()
	low_s = df["Low"].squeeze()
	vol_s = df["Volume"].squeeze()

	df["VWAP"] = TickerRegimeEngine.calculate_vwap(df)
	df["EMA_9"] = close_s.ewm(span=9, adjust=False).mean()
	df["EMA_21"] = close_s.ewm(span=21, adjust=False).mean()

	# Simulation State
	capital = initial_capital
	position_shares = 0
	entry_price = 0.0
	entry_time = None
	trades = []
	peak_price = 0.0

	# Iterate through bars
	for i in range(20, len(df)):
	sub_df = df.iloc[:i+1]
	regime_info = TickerRegimeEngine.analyze_ticker_regime(sub_df, "SNDK")

	current_time = df.index[i]
	bar_open = float(open_s.iloc[i])
	bar_high = float(high_s.iloc[i])
	bar_low = float(low_s.iloc[i])
	bar_close = float(close_s.iloc[i])
	bar_vwap = float(df["VWAP"].iloc[i])
	bar_ema9 = float(df["EMA_9"].iloc[i])
	bar_ema21 = float(df["EMA_21"].iloc[i])

	time_str = str(current_time)
	is_eod = "15:50" in time_str or "15:55" in time_str

	# 1. Check Exit if holding
	if position_shares > 0:
	peak_price = max(peak_price, bar_high)
	pnl_pct = (bar_close - entry_price) / entry_price * 100.0

	# Dynamic Trailing Exit for SNDK (GP Saggese wave rule: don't exit on micro dips)
	# Exit condition 1: Hard trend break (Close falls below EMA21 by > 0.6% AND below VWAP)
	# Exit condition 2: Large runner trailing stop (if up > 3.0%, trail 1.5% from peak)
	is_trend_broken = (bar_close < bar_ema21 * 0.994 and bar_close < bar_vwap)
	is_profit_trail_hit = (pnl_pct >= 2.5 and (peak_price - bar_close) / peak_price * 100.0 >= 1.5)

	if is_trend_broken or is_profit_trail_hit or is_eod:
	exit_price = bar_close
	trade_pnl = position_shares * (exit_price - entry_price)
	capital += position_shares * exit_price
	return_pct = (exit_price - entry_price) / entry_price * 100.0
	reason = "EOD Flatten" if is_eod else ("Trailing Stop Peak Lock" if is_profit_trail_hit else "Trend Life-Line Break")

	trades.append({
	"entry_time": entry_time,
	"exit_time": current_time,
	"side": "LONG",
	"shares": position_shares,
	"entry_price": entry_price,
	"exit_price": exit_price,
	"pnl_$": trade_pnl,
	"return_%": return_pct,
	"reason": reason
	})
	position_shares = 0
	entry_price = 0.0

	# 2. Check Entry if flat
	elif position_shares == 0 and not is_eod:
	# SNDK Entry Rule: In BULL_EXPANSION or strong upward momentum pullback to EMA9/VWAP
	if regime_info.regime == MarketRegime.BULL_EXPANSION:
	# Enter when price is comfortably above VWAP and testing EMA9
	if bar_close >= bar_vwap and bar_close >= bar_ema9 * 0.998:
	# Allocate 90% buying power into leader
	alloc_capital = capital * 0.90
	position_shares = int(alloc_capital / bar_close)
	if position_shares > 0:
	entry_price = bar_close
	entry_time = current_time
	peak_price = bar_close
	capital -= position_shares * bar_close

	# Close open position at end
	if position_shares > 0:
	final_close = float(close_s.iloc[-1])
	trade_pnl = position_shares * (final_close - entry_price)
	capital += position_shares * final_close
	return_pct = (final_close - entry_price) / entry_price * 100.0
	trades.append({
	"entry_time": entry_time,
	"exit_time": df.index[-1],
	"side": "LONG",
	"shares": position_shares,
	"entry_price": entry_price,
	"exit_price": final_close,
	"pnl_$": trade_pnl,
	"return_%": return_pct,
	"reason": "Market Close Final Mark"
	})

	# Summary Statistics
	df_trades = pd.DataFrame(trades)
	print("=" * 80)
	print(" 【SNDK 独立交易明细 (INDIVIDUAL TRADE LOG)】")
	print("=" * 80)
	if not df_trades.empty:
	for idx, row in df_trades.iterrows():
	win_icon = "🟢 盈利" if row["pnl_$"] > 0 else "🔴 亏损"
	print(f" Trade #{idx+1} \| {row['entry_time']} -> {row['exit_time']} \| "
	f"Entry: ${row['entry_price']:.2f} \| Exit: ${row['exit_price']:.2f} \| "
	f"PnL: ${row['pnl_$']:+,.2f} ({row['return_%']:+.2f}%) \| {win_icon} ({row['reason']})")
	else:
	print(" No trades triggered under strict regime criteria.")

	total_pnl = capital - initial_capital
	total_ret_pct = total_pnl / initial_capital * 100.0
	win_rate = (df_trades["pnl_$"] > 0).mean() * 100.0 if not df_trades.empty else 0.0

	print("\n" + "=" * 80)
	print(" 【SNDK 独立回测指标总结】")
	print("=" * 80)
	print(f" • 期末总资产 (Ending Equity): ${capital:,.2f}")
	print(f" • 累计总净利润 (Net Total PnL): ${total_pnl:+,.2f}")
	print(f" • 总收益率 (Total Net Return): {total_ret_pct:+.2f}%")
	print(f" • 胜率 (Win Rate): {win_rate:.1f}% ({len(df_trades)} 笔交易)")
	print("=" * 80 + "\n")

	if __name__ == "__main__":
	run_sndk_simulation()