import pandas as pd import numpy as np import gradio as gr import yfinance as yf # 使用標準的 yfinance 庫 import plotly.graph_objects as go # 定義偵測Top Wick且附加成交量篩選的Signal函式 def detect_top_wick_with_volume(df: pd.DataFrame, wick_ratio: float = 0.4, volume_multiplier: float = 1.3, min_range: float = 0.001) -> pd.Series: """ 偵測上影線且成交量放大的K線 Top Wick (上影線) 是一種看跌信號，表示價格在當天曾經上漲，但最終收盤價較低， 通常被視為賣壓增加的跡象，預示著可能的下跌趨勢。 """ body = np.abs(df['close'] - df['open']) upper_wick = df['high'] - df[['open', 'close']].max(axis=1) lower_wick = df[['open', 'close']].min(axis=1) - df['low'] total_range = df['high'] - df['low'] # 更嚴謹的處理小範圍 K 線 significant_range = total_range > min_range total_range = total_range.replace(0, 1e-6) avg_volume = df['volume'].rolling(window=20, min_periods=1).mean() condition = ( significant_range & (upper_wick / total_range > wick_ratio) & (upper_wick > lower_wick) & (df['volume'] > avg_volume * volume_multiplier) ) return condition.astype(int) # 定義交易策略函數 def top_wick_strategy(df, wick_ratio=0.4, volume_multiplier=1.3): """ 偵測上影線且成交量放大的策略 Top Wick 策略是一種做空策略，當出現上影線形態時，表示可能即將下跌， 此時應該做空（賣出），等待價格下跌後再買回平倉獲利。 """ signals = detect_top_wick_with_volume(df, wick_ratio, volume_multiplier) return signals # 定義出場策略函數 def exit_strategy(df, exit_bars=3, use_trailing_stop=False, stop_pct=0.02): """ 出場策略：固定K線數出場或追蹤止損 注意：這個函數只是生成可能的出場信號，實際出場會在交易模擬中處理 """ # 創建一個全為0的Series exit_signals = pd.Series(0, index=df.index) # 這裡不再使用固定位置的出場信號，而是在交易模擬中計算 # 如果啟用追蹤止損，生成止損信號 if use_trailing_stop: for i in range(1, len(df)): if df['close'].iloc[i] < df['close'].iloc[i-1] * (1 - stop_pct/100): exit_signals.iloc[i] = 1 return exit_signals # 定義回測與結果顯示流程 def run_backtest(symbol: str, start_date: str, end_date: str, wick_ratio: float, volume_multiplier: float, exit_bars: int = 3, use_trailing_stop: bool = False, stop_pct: float = 0.02, timeframe: str = "1d", trade_direction: str = "short"): try: print(f"Backtesting: {start_date} to {end_date}") # 使用 yfinance 獲取數據 ticker = yf.Ticker(symbol) # 檢查時間框架，如果是分鐘級數據且時間範圍過長，則提供警告 # yfinance 支持的時間框架 is_intraday = timeframe in ['1m', '2m', '5m', '15m', '30m', '60m', '90m'] if is_intraday: print(f"警告：yfinance 對分鐘級數據有限制，通常只能獲取最近 7-60 天的數據") # 如果是分鐘級數據，可以嘗試縮短時間範圍 from datetime import datetime, timedelta end_dt = datetime.strptime(end_date, "%Yn-%m-%d") start_dt = end_dt - timedelta(days=60) # 嘗試獲取最近 60 天的數據 adjusted_start = start_dt.strftime("%Y-%m-%d") print(f"自動調整時間範圍：{adjusted_start} 到 {end_date}") # 先嘗試用戶指定的時間範圍 df = ticker.history(start=start_date, end=end_date, interval=timeframe) # 如果數據為空，則嘗試調整後的時間範圍 if df.empty: print(f"使用調整後的時間範圍重試...") df = ticker.history(start=adjusted_start, end=end_date, interval=timeframe) else: # 日線或更長時間框架，直接使用指定的時間範圍 df = ticker.history(start=start_date, end=end_date, interval=timeframe) if df.empty: # 嘗試使用日線數據 if is_intraday: print(f"無法獲取 {symbol} 的分鐘級數據，嘗試使用日線數據...") df = ticker.history(start=start_date, end=end_date, interval="1d") if not df.empty: print(f"成功獲取 {symbol} 的日線數據，共 {len(df)} 條記錄") timeframe = "1d" # 更新時間框架 if df.empty: raise ValueError(f"無法獲取 {symbol} 的數據，請檢查股票代碼和日期範圍。美股分鐘級數據通常只有最近 7-60 天可用。") else: print(f"成功獲取 {symbol} 的 {timeframe} 數據，共 {len(df)} 條記錄，時間範圍：{df.index[0]} 到 {df.index[-1]}") # 確保列名符合我們的預期 df.columns = [col.lower() for col in df.columns] if 'adj close' in df.columns: df = df.rename(columns={'adj close': 'adj_close'}) # 計算信號 df['signal'] = top_wick_strategy(df, wick_ratio, volume_multiplier) df['exit'] = exit_strategy(df, exit_bars, use_trailing_stop, stop_pct) # 模擬交易 equity = 10000 # 初始資金 total_equity = 10000 # 累積權益 position = 0 # 持倉數量 (正數表示做多，負數表示做空) entry_price = 0 # 進場價格 entry_time = None # 進場時間 bars_since_entry = 0 # 自進場以來的K線數 equity_curve = pd.Series(index=df.index) signal_points = [] sell_points = [] # 先收集所有信號點，不考慮交易邏輯 for i, row in df.iterrows(): if row['signal'] == 1: signal_points.append({ 'timestamp': i, 'open': row['open'], 'high': row['high'], 'low': row['low'], 'close': row['close'], 'volume': row['volume'] }) # 再進行交易模擬 for i, row in df.iterrows(): # 如果持有倉位，增加持倉天數計數 if position != 0: bars_since_entry += 1 # 檢查是否應該出場 exit_signal = False exit_reason = "" if position > 0: # 做多倉位 # 固定K線數出場 if bars_since_entry >= exit_bars: exit_signal = True exit_reason = f"固定{exit_bars}天出場" # 追蹤止損出場 (對做多來說，價格下跌是止損) elif use_trailing_stop and row['close'] < entry_price * (1 - stop_pct/100): exit_signal = True exit_reason = "追蹤止損出場" elif position < 0: # 做空倉位 # 固定K線數出場 if bars_since_entry >= exit_bars: exit_signal = True exit_reason = f"固定{exit_bars}天出場" # 追蹤止損出場 (對做空來說，價格上漲是止損) elif use_trailing_stop and row['close'] > entry_price * (1 + stop_pct/100): exit_signal = True exit_reason = "追蹤止損出場" # 處理進出場 if position == 0 and row['signal'] == 1: if trade_direction == "long": # 做多 (買入) position = 100 # 假設買入 100 股 entry_price = row['close'] entry_time = i bars_since_entry = 0 else: # trade_direction == "short" # 做空 (賣出) position = -100 # 假設賣空 100 股 entry_price = row['close'] entry_time = i bars_since_entry = 0 elif position > 0 and exit_signal: # 做多倉位出場 # 出場 (賣出) # 計算本次交易盈虧 profit = position * (row['close'] - entry_price) # 做多盈利 = 賣出價 - 買入價 total_equity += profit position = 0 sell_points.append({ 'timestamp': i, 'open': row['open'], 'high': row['high'], 'low': row['low'], 'close': row['close'], 'volume': row['volume'], 'reason': exit_reason, 'profit': profit }) bars_since_entry = 0 elif position < 0 and exit_signal: # 做空倉位出場 # 出場 (買回) # 計算本次交易盈虧 profit = -position * (entry_price - row['close']) # 做空盈利 = 賣出價 - 買回價 total_equity += profit position = 0 sell_points.append({ 'timestamp': i, 'open': row['open'], 'high': row['high'], 'low': row['low'], 'close': row['close'], 'volume': row['volume'], 'reason': exit_reason, 'profit': profit }) bars_since_entry = 0 # 更新權益 if position > 0: # 做多倉位的當前價值 = 總權益 + 做多盈利 (當前價 - 買入價) current_equity = total_equity + position * (row['close'] - entry_price) elif position < 0: # 做空倉位的當前價值 = 總權益 + 做空盈利 (賣出價 - 當前價) current_equity = total_equity + (-position) * (entry_price - row['close']) else: current_equity = total_equity equity_curve[i] = current_equity # 輸出檢測到的信號數量 print(f"檢測到 {len(signal_points)} 個 Top Wick 信號") # 轉換為 DataFrame signal_points_df = pd.DataFrame(signal_points) if signal_points else pd.DataFrame() sell_points_df = pd.DataFrame(sell_points) if sell_points else pd.DataFrame() return equity_curve, signal_points_df, sell_points_df except Exception as e: print(f"回測過程中發生錯誤: {str(e)}") raise # 畫圖：含 Top Wick 和 Sell 標記的 Plotly 圖 def plot_equity_with_signals(equity_curve: pd.Series, signal_points: pd.DataFrame, sell_points: pd.DataFrame) -> go.Figure: fig = go.Figure() fig.add_trace(go.Scatter( x=equity_curve.index, y=equity_curve.values, mode='lines', name='Equity Curve' )) if not signal_points.empty: # 確保時間戳存在於 equity_curve 中 valid_signals = signal_points[signal_points['timestamp'].isin(equity_curve.index)] if not valid_signals.empty: fig.add_trace(go.Scatter( x=valid_signals['timestamp'], y=[equity_curve.loc[ts] for ts in valid_signals['timestamp']], mode='markers', marker=dict(color='red', size=8, symbol='x'), name='Top Wick Signal' )) if not sell_points.empty: # 確保時間戳存在於 equity_curve 中 valid_sells = sell_points[sell_points['timestamp'].isin(equity_curve.index)] if not valid_sells.empty: fig.add_trace(go.Scatter( x=valid_sells['timestamp'], y=[equity_curve.loc[ts] for ts in valid_sells['timestamp']], mode='markers', marker=dict(color='green', size=8, symbol='circle'), name='Sell Point' )) fig.update_layout(title='Equity Curve with Signal & Sell Points', xaxis_title='Time', yaxis_title='Equity') return fig # Gradio介面 def gradio_interface(symbol, start_date, end_date, wick_ratio, volume_multiplier, exit_bars=3, use_trailing_stop=False, stop_pct=0.02, timeframe="1d", trade_direction="short"): try: # 驗證輸入 if not symbol or not start_date or not end_date: # 創建一個空的圖表而不是返回字符串 empty_fig = go.Figure() empty_fig.add_annotation( text="請填寫所有必要欄位", xref="paper", yref="paper", x=0.5, y=0.5, showarrow=False, font=dict(size=20) ) return empty_fig, pd.DataFrame(), "回測失敗" # 執行回測 equity_curve, signal_points, sell_points = run_backtest( symbol, start_date, end_date, wick_ratio, volume_multiplier, exit_bars, use_trailing_stop, stop_pct, timeframe, trade_direction ) # 計算回測統計數據 total_signals = len(signal_points) total_trades = len(sell_points) if len(equity_curve) > 0: final_return = (equity_curve.iloc[-1] / equity_curve.iloc[0] - 1) * 100 max_drawdown = ((equity_curve / equity_curve.cummax()) - 1).min() * 100 else: final_return = 0 max_drawdown = 0 stats = f"總信號數: {total_signals}, 總交易數: {total_trades}, 總回報: {final_return:.2f}%, 最大回撤: {max_drawdown:.2f}%" # 繪製圖表 fig = plot_equity_with_signals(equity_curve, signal_points, sell_points) # 準備表格數據 (同時顯示進場和出場點) if not signal_points.empty: signal_points = signal_points.copy() signal_points['類型'] = '進場信號' if not sell_points.empty: sell_points = sell_points.copy() sell_points['類型'] = sell_points.apply( lambda x: f"出場信號 ({x.get('reason', '')}, 盈利: {x.get('profit', 0):.2f})" if 'reason' in x else '出場信號', axis=1) # 合併表格 all_points = pd.concat([signal_points, sell_points]).sort_values('timestamp') if not (signal_points.empty and sell_points.empty) else pd.DataFrame() table_data = all_points[['timestamp', 'open', 'high', 'low', 'close', 'volume', '類型']] if not all_points.empty else pd.DataFrame() return fig, table_data, stats except Exception as e: import traceback error_details = traceback.format_exc() # 創建一個包含錯誤信息的圖表 error_fig = go.Figure() error_fig.add_annotation( text=f"發生錯誤: {str(e)}", xref="paper", yref="paper", x=0.5, y=0.5, showarrow=False, font=dict(size=14, color="red") ) return error_fig, pd.DataFrame(), f"回測失敗: {str(e)}" with gr.Blocks() as demo: gr.Markdown(""" # Top Wick + Volume 偵測器 (含進出場標記) 請輸入參數並開始回測！ """) with gr.Tab("基本設定"): symbol = gr.Textbox(label="股票代碼 (如: AAPL, 2330.TW)", value="AAPL") start_date = gr.Textbox(label="開始日期 (yyyy-mm-dd)", value="2025-01-01") end_date = gr.Textbox(label="結束日期 (yyyy-mm-dd)", value="2025-04-28") timeframe = gr.Dropdown( label="時間框架", choices=["1d", "5d", "1wk", "1mo", "3mo"], value="1d", info="Top Wick 策略在日線及以上時間框架效果最佳" ) with gr.Tab("進場設定"): wick_ratio = gr.Slider(label="上影線比例門檻", minimum=0.1, maximum=1.0, step=0.05, value=0.4, info="業界常用範圍：0.3-0.5，較低值檢測更多信號") volume_multiplier = gr.Slider(label="成交量大於過去平均倍數", minimum=0.5, maximum=5.0, step=0.1, value=1.3, info="業界常用範圍：1.2-1.5") trade_direction = gr.Radio( label="交易方向", choices=["long", "short"], value="short", info="long: 做多(買入)，short: 做空(賣出)。Top Wick 通常是看跌信號，適合做空。" ) with gr.Tab("出場設定"): exit_bars = gr.Slider(label="固定K線數出場", minimum=1, maximum=20, step=1, value=3, info="業界常用範圍：3-5 個交易日，進場後經過這麼多天就出場") use_trailing_stop = gr.Checkbox(label="啟用追蹤止損", value=True, info="啟用後，當價格下跌超過設定百分比時會觸發出場") stop_pct = gr.Slider(label="止損百分比", minimum=0.5, maximum=10.0, step=0.5, value=2.0, info="業界常用範圍：1-3%，價格下跌超過這個百分比時出場") # 當啟用追蹤止損時顯示止損百分比 use_trailing_stop.change( fn=lambda x: gr.update(visible=x), inputs=[use_trailing_stop], outputs=[stop_pct] ) output_plot = gr.Plot() output_table = gr.Dataframe() output_stats = gr.Textbox(label="回測統計") btn = gr.Button("開始回測") btn.click( fn=gradio_interface, inputs=[symbol, start_date, end_date, wick_ratio, volume_multiplier, exit_bars, use_trailing_stop, stop_pct, timeframe, trade_direction], outputs=[output_plot, output_table, output_stats] ) demo.launch()