如何構建屬於你自己的交易系統

1. 為什麼你需要一套交易系統？

很多交易新手每天盯盤，憑「感覺」買賣，結果不斷追漲殺跌，賬戶資金越來越少。問題的根源不在於選錯了股票，而在於——你沒有一套明確的交易系統。

一套完整的交易系統能幫你解決三個核心問題：

• 消除情緒干擾：買賣決策由規則驅動，而非恐懼或貪婪
• 保證一致性：每一筆交易都遵循相同的邏輯，結果可統計、可復現
• 可迭代優化：有了明確規則，才能回測、量化評估、不斷改進

「交易系統不是預測市場，而是在不確定的市場中，用確定的規則持續執行有正期望值的策略。」

簡而言之，交易系統就是一份完整的操作手冊——它明確告訴你什麼時候買、買多少、什麼時候賣、虧損多少該止損。沒有系統的交易者就像沒有地圖的旅行者，或許偶爾走對方向，卻永遠無法穩定到達目的地。

2. 交易系統的五大核心模塊

一個完整的交易系統通常由以下五個模塊組成：

這五個模塊環環相扣，缺一不可。很多新手只關注「信號」（什麼時候買），但忽略了倉位管理和風險控制——這恰恰是職業交易者最重視的部分。

3. 信號生成：何時買、何時賣？

信號生成模塊負責回答最基本的問題：什麼條件滿足時開倉？什麼條件滿足時平倉？

3.1 常見的信號類型

3.2 示例：雙均線交叉策略

以最經典的雙均線交叉策略為例，規則非常簡單：

• 買入信號：短期均線（如 10日）從下方穿越長期均線（如 50日）——「金叉」
• 賣出信號：短期均線從上方穿越長期均線——「死叉」

import pandas as pd
import numpy as np

class DualSMASignal:
    """雙均線交叉信號生成器"""
    
    def __init__(self, short_window=10, long_window=50):
        self.short_window = short_window
        self.long_window = long_window
    
    def generate(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        生成交易信號
        
        Parameters:
            df: 包含 'close' 列的 DataFrame
            
        Returns:
            新增 'signal' 列的 DataFrame
            signal = 1  → 買入
            signal = -1 → 賣出
            signal = 0  → 持有/觀望
        """
        df = df.copy()
        
        # 計算短期和長期均線
        df['sma_short'] = df['close'].rolling(window=self.short_window).mean()
        df['sma_long'] = df['close'].rolling(window=self.long_window).mean()
        
        # 生成信號
        df['signal'] = 0
        df.loc[df['sma_short'] > df['sma_long'], 'signal'] = 1   # 短均線在上 → 多頭
        df.loc[df['sma_short'] < df['sma_long'], 'signal'] = -1  # 短均線在下 → 空頭
        
        # 只在信號變化時觸發交易（去除重複信號）
        df['trade'] = df['signal'].diff().fillna(0)
        # trade = 2  → 金叉（從 -1 變 1），買入
        # trade = -2 → 死叉（從 1 變 -1），賣出
        
        return df


# 使用示例
if __name__ == "__main__":
    # 假設你已經有了 SPY 的歷史數據
    df = pd.read_csv("spy_daily.csv", parse_dates=['date'])
    
    signal_gen = DualSMASignal(short_window=10, long_window=50)
    df = signal_gen.generate(df)
    
    # 統計交易次數
    buy_signals = (df['trade'] == 2).sum()
    sell_signals = (df['trade'] == -2).sum()
    print(f"買入信號: {buy_signals} 次")
    print(f"賣出信號: {sell_signals} 次")

3.3 信號生成的設計原則

1. 規則必須明確：能被寫成代碼的規則才是好規則，模糊的「感覺」不算
2. 避免過度擬合：參數越少越好，規則越簡單越穩健
3. 考慮交易成本：信號太頻繁，手續費和滑點會吃掉利潤
4. 進場重要，出場更重要：很多新手只研究買入，卻沒有明確的賣出規則

4. 倉位管理：每次該買多少？

即使你的信號系統 100% 準確（當然這不可能），糟糕的倉位管理也能讓你破產。倉位管理回答的是：每筆交易投入多少資金？

4.1 常見的倉位管理方法

4.2 推薦：固定風險法（Fixed Fractional）

對於大多數交易者，固定風險法是最實用的方案。核心思想是：每筆交易最多虧損總資金的一個固定百分比（通常 1%-2%）。

class FixedRiskPositionSizer:
    """固定風險倉位管理器"""
    
    def __init__(self, risk_per_trade=0.02):
        """
        Parameters:
            risk_per_trade: 每筆交易的最大風險比例（默認 2%）
        """
        self.risk_per_trade = risk_per_trade
    
    def calculate_position_size(
        self, 
        total_capital: float,
        entry_price: float,
        stop_loss_price: float
    ) -> int:
        """
        計算倉位大小
        
        Parameters:
            total_capital: 總資金
            entry_price: 計劃入場價格
            stop_loss_price: 止損價格
            
        Returns:
            建議買入的股數
        """
        # 每股風險 = 入場價 - 止損價
        risk_per_share = abs(entry_price - stop_loss_price)
        
        if risk_per_share == 0:
            return 0
        
        # 最大可承受風險金額
        max_risk_amount = total_capital * self.risk_per_trade
        
        # 倉位大小 = 最大風險金額 / 每股風險
        position_size = int(max_risk_amount / risk_per_share)
        
        # 確保不超過總資金
        max_shares = int(total_capital / entry_price)
        position_size = min(position_size, max_shares)
        
        return position_size


# 使用示例
sizer = FixedRiskPositionSizer(risk_per_trade=0.02)

# 場景：總資金 100,000，計劃買入價 150，止損價 145
shares = sizer.calculate_position_size(
    total_capital=100000,
    entry_price=150,
    stop_loss_price=145
)
print(f"建議買入: {shares} 股")
print(f"投入資金: ${shares * 150:,.0f}")
print(f"最大虧損: ${shares * 5:,.0f} ({shares * 5 / 100000:.1%} of total)")
# 輸出: 建議買入: 400 股, 投入資金: $60,000, 最大虧損: $2,000 (2.0%)

5. 風險控制：活下來比賺錢更重要

華爾街有一句老話：「Rule #1: Don't lose money. Rule #2: Don't forget Rule #1.」風險控制的目標不是讓你不虧錢，而是確保單筆虧損和累計虧損都在可控範圍內，讓你有足夠的「子彈」等到盈利的機會出現。

5.1 三層風控體系

5.2 止損策略的實現

class RiskManager:
    """三層風控管理器"""
    
    def __init__(
        self,
        max_risk_per_trade=0.02,   # 單筆最大風險 2%
        max_daily_loss=0.05,       # 日度最大虧損 5%
        max_drawdown=0.20          # 最大回撤 20%
    ):
        self.max_risk_per_trade = max_risk_per_trade
        self.max_daily_loss = max_daily_loss
        self.max_drawdown = max_drawdown
        
        self.peak_capital = 0       # 歷史最高資金
        self.daily_start_capital = 0  # 當日起始資金
    
    def update_daily(self, current_capital: float):
        """每日開盤前更新"""
        self.daily_start_capital = current_capital
        self.peak_capital = max(self.peak_capital, current_capital)
    
    def can_trade(self, current_capital: float) -> dict:
        """
        檢查是否允許交易
        
        Returns:
            {'allowed': bool, 'reason': str}
        """
        # 檢查日度止損
        daily_pnl = (current_capital - self.daily_start_capital) / self.daily_start_capital
        if daily_pnl <= -self.max_daily_loss:
            return {
                'allowed': False,
                'reason': f'日度虧損已達 {daily_pnl:.1%}，超過限額 {-self.max_daily_loss:.1%}'
            }
        
        # 檢查最大回撤
        drawdown = (self.peak_capital - current_capital) / self.peak_capital
        if drawdown >= self.max_drawdown:
            return {
                'allowed': False,
                'reason': f'回撤已達 {drawdown:.1%}，超過限額 {self.max_drawdown:.1%}'
            }
        
        return {'allowed': True, 'reason': '風控通過'}
    
    def calculate_stop_loss(
        self,
        entry_price: float,
        atr: float,
        multiplier: float = 2.0
    ) -> float:
        """
        基於 ATR 計算止損價格
        
        Parameters:
            entry_price: 入場價格
            atr: 當前 ATR 值
            multiplier: ATR 倍數（默認 2 倍）
            
        Returns:
            止損價格
        """
        stop_loss = entry_price - (atr * multiplier)
        return round(stop_loss, 2)


# 使用示例
rm = RiskManager(
    max_risk_per_trade=0.02,
    max_daily_loss=0.05,
    max_drawdown=0.20
)

# 每日更新
rm.update_daily(current_capital=100000)

# 交易前檢查
result = rm.can_trade(current_capital=96000)
print(result)
# {'allowed': True, 'reason': '風控通過'}

result = rm.can_trade(current_capital=94500)
print(result)
# {'allowed': False, 'reason': '日度虧損已達 -5.5%，超過限額 -5.0%'}

6. 回測驗證：用歷史數據檢驗你的系統

有了信號、倉位管理和風控規則之後，下一步就是用歷史數據來驗證你的系統——這就是回測（Backtesting）。回測的本質是：假設你在過去的某個時間段使用這套系統交易，結果會怎樣？

6.1 回測的核心指標

6.2 用 Backtrader 實現簡單回測

import backtrader as bt

class DualSMAStrategy(bt.Strategy):
    """雙均線交叉策略"""
    
    params = (
        ('short_period', 10),
        ('long_period', 50),
        ('risk_per_trade', 0.02),
    )
    
    def __init__(self):
        # 計算均線
        self.sma_short = bt.indicators.SMA(
            self.data.close, period=self.params.short_period
        )
        self.sma_long = bt.indicators.SMA(
            self.data.close, period=self.params.long_period
        )
        # 均線交叉信號
        self.crossover = bt.indicators.CrossOver(
            self.sma_short, self.sma_long
        )
    
    def next(self):
        if not self.position:
            # 無持倉時，金叉買入
            if self.crossover > 0:
                # 使用固定風險法計算倉位
                risk_amount = self.broker.getvalue() * self.params.risk_per_trade
                atr = self.data.close[0] * 0.02  # 簡化：用價格的2%作為止損距離
                size = int(risk_amount / atr) if atr > 0 else 0
                if size > 0:
                    self.buy(size=size)
                    print(f'📈 買入 {size} 股 @ {self.data.close[0]:.2f}')
        else:
            # 有持倉時，死叉賣出
            if self.crossover < 0:
                self.close()
                print(f'📉 賣出 @ {self.data.close[0]:.2f}')


def run_backtest():
    """運行回測"""
    cerebro = bt.Cerebro()
    
    # 添加策略
    cerebro.addstrategy(DualSMAStrategy)
    
    # 添加數據
    data = bt.feeds.GenericCSVData(
        dataname='spy_daily.csv',
        dtformat='%Y-%m-%d',
        datetime=0,
        open=1, high=2, low=3, close=4, volume=5,
        openinterest=-1
    )
    cerebro.adddata(data)
    
    # 設置初始資金
    cerebro.broker.setcash(100000.0)
    
    # 設置手續費
    cerebro.broker.setcommission(commission=0.001)
    
    # 添加分析器
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
    cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
    cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
    
    print(f'起始資金: ${cerebro.broker.getvalue():,.2f}')
    
    # 運行回測
    results = cerebro.run()
    strat = results[0]
    
    print(f'最終資金: ${cerebro.broker.getvalue():,.2f}')
    print(f'總收益率: {cerebro.broker.getvalue() / 100000 - 1:.2%}')
    print(f'夏普比率: {strat.analyzers.sharpe.get_analysis().get("sharperatio", "N/A")}')
    print(f'最大回撤: {strat.analyzers.drawdown.get_analysis().max.drawdown:.2f}%')
    
    return cerebro


if __name__ == "__main__":
    cerebro = run_backtest()
    cerebro.plot()  # 繪製回測圖表

6.3 回測中的常見陷阱

回測結果「看起來很美」但實盤虧損？可能掉進了這些坑：

1. 前視偏誤（Look-Ahead Bias）：使用了未來數據。例如用收盤價來做「開盤時」的決策
2. 過擬合（Overfitting）：參數過多，完美匹配歷史但無法預測未來
3. 忽略交易成本：手續費、滑點、衝擊成本，在高頻策略中尤為致命
4. 生存偏誤（Survivorship Bias）：只測試「活下來」的股票，忽略了退市的個股

7. 從回測到實盤：部署你的系統

回測結果令人滿意之後，就可以考慮將系統部署到實盤環境。但從回測到實盤之間，還有一個關鍵環節——模擬盤測試（Paper Trading）。

7.1 上線路線圖

7.2 實盤部署的技術架構

一個完整的自動化交易系統通常包含以下組件：

"""
交易系統架構示意

┌─────────────────────────────────────────────┐
│                 調度器 (Scheduler)            │
│          定時觸發 / 事件驅動                    │
└──────────────┬──────────────────────────────┘
               │
    ┌──────────▼──────────┐
    │    數據引擎 (Data)    │  ← 行情API (YFinance / Longport)
    │  獲取最新行情數據      │
    └──────────┬──────────┘
               │
    ┌──────────▼──────────┐
    │  信號引擎 (Signal)    │  ← 技術指標計算
    │  生成交易信號          │
    └──────────┬──────────┘
               │
    ┌──────────▼──────────┐
    │  風控引擎 (Risk)      │  ← 倉位計算 + 風控檢查
    │  倉位管理 & 風險檢查    │
    └──────────┬──────────┘
               │
    ┌──────────▼──────────┐
    │  執行引擎 (Executor)  │  → 券商API (Longport / IB)
    │  下單 / 撤單 / 狀態    │
    └──────────┬──────────┘
               │
    ┌──────────▼──────────┐
    │  日誌 & 監控           │  → 記錄交易 / 發送通知
    │  (Logging & Alert)   │
    └─────────────────────┘
"""

7.3 自動化交易的關鍵注意事項

1. 網絡穩定性：使用雲服務器（如 AWS、阿里雲），而非本地電腦
2. 異常處理：API 斷線、下單失敗、數據缺失都要有兜底方案
3. 日誌記錄：每一筆交易、每一個信號都要記錄，方便事後排查
4. 監控告警：異常情況（如大幅虧損、API 斷線）要即時通知
5. 資金安全：設置 API 權限限制，避免程序 Bug 導致意外大額交易

8. 總結：構建系統的路線圖

讓我們回顧一下構建完整交易系統的路線：

「優秀的交易系統不是一天建成的。它是在實踐 → 復盤 → 改進的循環中，一步步打磨出來的。最重要的是——開始動手。」

延伸閱讀

構建交易系統需要多方面的知識支撐，以下是相關的進階內容：

• AKShare / YFinance / Longport 數據下載教程 — 獲取歷史數據
• AI 交互構建 SPY 雙均線策略回測 — 動手實現你的第一個回測
• 回測陷阱：新手最容易踩的7個坑 — 避開回測中的常見錯誤
• AI 構建 Longport SDK SPY 交易策略 — 從回測到實盤全流程
• 量化交易的不可能三角 — 理解交易系統的根本約束

想要系統學習完整的量化交易技能？歡迎了解我的 10節實戰課程！