如何和 AI 交互構建對接 Longport SDK 的 SPY 實盤交易策略

1. 為什麼選擇 Longport SDK？

在上一篇文章中，我們用 AI 構建了一個基於 Backtrader 的雙均線策略回測程序。但回測終歸是回測——真正能賺錢的系統，必須能連接券商、獲取實時行情、自動下單。

Longport（長橋）是目前最適合量化交易者的港美股券商之一，其 Python SDK 提供了：

行情 API：實時/歷史 K 線、逐筆成交、盤口深度
交易 API：下單、撤單、查詢持倉和訂單
推送功能：WebSocket 實時行情推送
完善文檔：官方 Python SDK 文檔清晰，社區活躍

💡 本文目標：

通過與 AI 對話，從零構建一個對接 Longport SDK 的 SPY 均線突破交易策略，涵蓋數據獲取、策略邏輯、實盤下單、風控監控全流程。

對比項	Backtrader 回測	Longport 實盤
數據來源	CSV 歷史數據	實時行情 API
執行方式	模擬交易引擎	真實券商下單
滑點/延遲	假設為 0	真實市場延遲
資金管理	虛擬資金	真金白銀

2. 環境準備與 API 配置

2.1 安裝 Longport SDK

pip install longport

2.2 獲取 API 密鑰

註冊 Longport 長橋賬戶
前往 開發者中心 → API 管理
創建 API Key，獲得三個關鍵信息：
- LONGPORT_APP_KEY
- LONGPORT_APP_SECRET
- LONGPORT_ACCESS_TOKEN

2.3 配置環境變量

# 在 ~/.bashrc 或 ~/.zshrc 中添加
export LONGPORT_APP_KEY="你的 App Key"
export LONGPORT_APP_SECRET="你的 App Secret"
export LONGPORT_ACCESS_TOKEN="你的 Access Token"

# 生效
source ~/.zshrc

⚠️ 安全提醒：

永遠不要把 API 密鑰硬編碼在代碼裡或提交到 Git！使用環境變量是最佳實踐。

3. 第一步：用 AI 獲取 SPY 實時行情數據

Prompt #1：通過 Longport SDK 獲取 SPY 行情

📝 複製這段 Prompt

我已經安裝了 longport Python SDK，並配置好了環境變量（LONGPORT_APP_KEY、LONGPORT_APP_SECRET、LONGPORT_ACCESS_TOKEN）。現在我要獲取 SPY.US 的歷史日 K 線數據和實時報價。要求： 1. 使用 longport.openapi 中的 QuoteContext 2. 從環境變量自動讀取配置（使用 Config.from_env()） 3. 獲取最近 120 個交易日的日 K 線（用於計算均線） 4. 獲取當前實時報價（最新價、漲跌幅） 5. 將 K 線數據轉換為 pandas DataFrame 方便後續處理 6. 打印數據預覽請提供完整可運行的 Python 代碼。

AI 生成的代碼

from longport.openapi import QuoteContext, Config, Period, AdjustType
import pandas as pd
from datetime import datetime

def get_spy_quote_data():
    """通過 Longport SDK 獲取 SPY.US 行情數據"""
    
    # 從環境變量讀取配置
    config = Config.from_env()
    ctx = QuoteContext(config)
    
    symbol = "SPY.US"
    
    # ===== 1. 獲取實時報價 =====
    print("=" * 60)
    print("SPY.US 實時報價")
    print("=" * 60)
    
    quote = ctx.quote([symbol])
    if quote:
        q = quote[0]
        print(f"股票代碼:   {symbol}")
        print(f"最新價格:   ${q.last_done}")
        print(f"昨收價格:   ${q.prev_close}")
        print(f"今日最高:   ${q.high}")
        print(f"今日最低:   ${q.low}")
        print(f"成交量:     {q.volume:,}")
        print(f"成交額:     ${q.turnover:,.2f}")
        
        change = float(q.last_done) - float(q.prev_close)
        change_pct = change / float(q.prev_close) * 100
        emoji = "🟢" if change >= 0 else "🔴"
        print(f"漲跌幅:     {emoji} {change_pct:+.2f}%")
    
    # ===== 2. 獲取日 K 線數據 =====
    print(f"\n{'=' * 60}")
    print(f"SPY.US 日 K 線數據（最近 120 個交易日）")
    print(f"{'=' * 60}")
    
    candlesticks = ctx.candlesticks(
        symbol=symbol,
        period=Period.Day,
        count=120,
        adjust_type=AdjustType.ForwardAdjust
    )
    
    # 轉換為 DataFrame
    records = []
    for candle in candlesticks:
        records.append({
            'date': candle.timestamp.strftime('%Y-%m-%d'),
            'open': float(candle.open),
            'high': float(candle.high),
            'low': float(candle.low),
            'close': float(candle.close),
            'volume': int(candle.volume),
            'turnover': float(candle.turnover),
        })
    
    df = pd.DataFrame(records)
    df['date'] = pd.to_datetime(df['date'])
    df = df.sort_values('date').reset_index(drop=True)
    
    print(f"\n共獲取 {len(df)} 條 K 線數據")
    print(f"日期範圍: {df['date'].iloc[0].strftime('%Y-%m-%d')} 至 "
          f"{df['date'].iloc[-1].strftime('%Y-%m-%d')}")
    print(f"\n最新 5 條數據:")
    print(df.tail().to_string(index=False))
    
    return df, ctx


if __name__ == "__main__":
    df, ctx = get_spy_quote_data()

✅ 運行結果：

成功獲取 SPY.US 的實時報價和 120 條日 K 線數據，已轉換為 DataFrame 格式。

💡 與回測版的區別：

回測版用 yfinance 下載 CSV，而這裡直接通過 Longport 的行情接口實時獲取數據，數據更準確、延遲更低。

4. 第二步：用 AI 構建均線突破策略邏輯

Prompt #2：實現策略信號計算

📝 複製這段 Prompt

我已經通過 Longport SDK 獲取了 SPY.US 的日 K 線 DataFrame，列名為 date, open, high, low, close, volume。現在請幫我構建一個均線突破策略（Moving Average Crossover）的信號計算模塊。策略邏輯： - 快線：20 日 EMA（指數移動平均線） - 慢線：60 日 EMA - 當快線從下方向上穿越慢線（金叉），生成買入信號 - 當快線從上方向下穿越慢線（死叉），生成賣出信號 - 額外過濾條件：只有當收盤價在 200 日均線之上時，才執行買入（趨勢過濾）要求： 1. 封裝成一個 Strategy 類 2. 提供 calculate_signals(df) 方法，返回帶有信號列的 DataFrame 3. 提供 get_latest_signal() 方法，返回當前最新信號（"BUY" / "SELL" / "HOLD"） 4. 使用 pandas 計算均線，不依賴 TA-Lib 等外部庫 5. 代碼結構清晰，有詳細註釋請提供完整代碼。

AI 生成的代碼

import pandas as pd
import numpy as np

class EMACrossoverStrategy:
    """
    EMA 均線突破策略
    
    - 快線: 20 日 EMA
    - 慢線: 60 日 EMA
    - 趨勢過濾: 200 日 SMA
    
    買入條件: 快線金叉慢線 + 收盤價在 200 日均線之上
    賣出條件: 快線死叉慢線
    """
    
    def __init__(self, fast_period=20, slow_period=60, trend_period=200):
        self.fast_period = fast_period
        self.slow_period = slow_period
        self.trend_period = trend_period
        self.signals_df = None
    
    def calculate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        計算交易信號
        
        Parameters:
            df: 包含 date, open, high, low, close, volume 列的 DataFrame
            
        Returns:
            帶有信號列的 DataFrame
        """
        df = df.copy()
        
        # ===== 計算均線 =====
        df['ema_fast'] = df['close'].ewm(span=self.fast_period, adjust=False).mean()
        df['ema_slow'] = df['close'].ewm(span=self.slow_period, adjust=False).mean()
        df['sma_trend'] = df['close'].rolling(window=self.trend_period).mean()
        
        # ===== 判斷交叉 =====
        # 快線與慢線的差值
        df['ema_diff'] = df['ema_fast'] - df['ema_slow']
        df['ema_diff_prev'] = df['ema_diff'].shift(1)
        
        # 金叉：前一天差值 <= 0，今天差值 > 0
        df['golden_cross'] = (df['ema_diff_prev'] <= 0) & (df['ema_diff'] > 0)
        
        # 死叉：前一天差值 >= 0，今天差值 < 0
        df['death_cross'] = (df['ema_diff_prev'] >= 0) & (df['ema_diff'] < 0)
        
        # ===== 趨勢過濾 =====
        # 只有收盤價在 200 日均線之上才允許買入
        df['above_trend'] = df['close'] > df['sma_trend']
        
        # ===== 生成交易信號 =====
        df['signal'] = 'HOLD'
        
        # 買入信號: 金叉 + 趨勢向上
        df.loc[df['golden_cross'] & df['above_trend'], 'signal'] = 'BUY'
        
        # 賣出信號: 死叉（不需要趨勢過濾）
        df.loc[df['death_cross'], 'signal'] = 'SELL'
        
        self.signals_df = df
        return df
    
    def get_latest_signal(self) -> dict:
        """
        獲取最新交易信號
        
        Returns:
            dict: {
                'signal': 'BUY' / 'SELL' / 'HOLD',
                'date': 日期,
                'close': 收盤價,
                'ema_fast': 快線值,
                'ema_slow': 慢線值,
                'sma_trend': 趨勢線值,
                'above_trend': 是否在趨勢線之上
            }
        """
        if self.signals_df is None:
            raise ValueError("請先調用 calculate_signals() 計算信號")
        
        latest = self.signals_df.iloc[-1]
        
        return {
            'signal': latest['signal'],
            'date': latest['date'],
            'close': latest['close'],
            'ema_fast': round(latest['ema_fast'], 2),
            'ema_slow': round(latest['ema_slow'], 2),
            'sma_trend': round(latest['sma_trend'], 2) if pd.notna(latest['sma_trend']) else None,
            'above_trend': latest['above_trend'],
        }
    
    def print_signal_summary(self):
        """打印最新信號摘要"""
        info = self.get_latest_signal()
        
        signal_emoji = {'BUY': '🟢 買入', 'SELL': '🔴 賣出', 'HOLD': '⏸️ 持有'}
        
        print(f"\n{'=' * 50}")
        print(f"📊 策略信號摘要")
        print(f"{'=' * 50}")
        print(f"日期:       {info['date']}")
        print(f"收盤價:     ${info['close']:.2f}")
        print(f"EMA{self.fast_period}:     ${info['ema_fast']}")
        print(f"EMA{self.slow_period}:     ${info['ema_slow']}")
        print(f"SMA{self.trend_period}:    ${info['sma_trend']}")
        print(f"趨勢過濾:   {'✅ 趨勢向上' if info['above_trend'] else '❌ 趨勢向下'}")
        print(f"{'=' * 50}")
        print(f"當前信號:   {signal_emoji.get(info['signal'], info['signal'])}")
        print(f"{'=' * 50}")


# ===== 使用示例 =====
if __name__ == "__main__":
    # 假設 df 已通過 Longport SDK 獲取
    from longport.openapi import QuoteContext, Config, Period, AdjustType
    
    config = Config.from_env()
    ctx = QuoteContext(config)
    
    # 獲取 200+ 條 K 線用於計算 200 日均線
    candlesticks = ctx.candlesticks(
        symbol="SPY.US",
        period=Period.Day,
        count=300,
        adjust_type=AdjustType.ForwardAdjust
    )
    
    records = [{
        'date': c.timestamp.strftime('%Y-%m-%d'),
        'open': float(c.open),
        'high': float(c.high),
        'low': float(c.low),
        'close': float(c.close),
        'volume': int(c.volume),
    } for c in candlesticks]
    
    df = pd.DataFrame(records)
    df['date'] = pd.to_datetime(df['date'])
    df = df.sort_values('date').reset_index(drop=True)
    
    # 計算信號
    strategy = EMACrossoverStrategy(fast_period=20, slow_period=60, trend_period=200)
    signals_df = strategy.calculate_signals(df)
    strategy.print_signal_summary()
    
    # 查看最近的交易信號
    recent_signals = signals_df[signals_df['signal'] != 'HOLD'].tail(5)
    print(f"\n最近 5 次交易信號:")
    print(recent_signals[['date', 'close', 'ema_fast', 'ema_slow', 'signal']].to_string(index=False))

✅ 策略邏輯亮點：

相比回測版的雙 SMA 策略，這裡升級為 EMA + 趨勢過濾：EMA 反應更靈敏，200 日均線過濾能避免在下跌趨勢中逆勢做多。

5. 第三步：用 AI 實現實盤下單模塊

Prompt #3：Longport 下單與持倉管理

📝 複製這段 Prompt

現在我需要通過 Longport SDK 實現自動下單功能。要求： 1. 使用 longport.openapi 中的 TradeContext 2. 實現以下功能： - 查詢當前賬戶餘額和持倉 - 提交市價買入訂單（指定股數） - 提交市價賣出訂單（全部持倉） - 查詢訂單狀態 - 撤銷未成交訂單 3. 封裝成 TradeExecutor 類 4. 加入基本的錯誤處理和日誌輸出 5. 支持「模擬」模式（只打印不實際下單），方便測試注意：SPY.US 是美股，市價單的 orderType 使用 OrderType.MO 請提供完整代碼。

AI 生成的代碼

from longport.openapi import (
    TradeContext, Config, OrderSide, OrderType,
    TimeInForceType, OutsideRTH
)
from decimal import Decimal
from datetime import datetime
import time
import logging

# 配置日誌
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)


class TradeExecutor:
    """
    Longport 交易執行器
    
    支持：查詢賬戶、下單、撤單、持倉管理
    """
    
    def __init__(self, simulate=True):
        """
        初始化交易執行器
        
        Parameters:
            simulate: 是否為模擬模式（True = 只打印不下單）
        """
        self.simulate = simulate
        self.config = Config.from_env()
        self.ctx = TradeContext(self.config)
        
        mode = "🧪 模擬模式" if simulate else "🔴 實盤模式"
        logger.info(f"交易執行器已初始化 | {mode}")
    
    def get_account_balance(self) -> dict:
        """查詢賬戶餘額"""
        balances = self.ctx.account_balance()
        
        for balance in balances:
            for cash_info in balance.cash_infos:
                if cash_info.currency == "USD":
                    info = {
                        'currency': 'USD',
                        'available': float(cash_info.available),
                        'frozen': float(cash_info.frozen),
                        'total': float(cash_info.total_cash),
                    }
                    logger.info(
                        f"賬戶餘額 | 可用: ${info['available']:,.2f} | "
                        f"凍結: ${info['frozen']:,.2f} | "
                        f"總計: ${info['total']:,.2f}"
                    )
                    return info
        
        logger.warning("未找到 USD 賬戶餘額")
        return {}
    
    def get_positions(self, symbol="SPY.US") -> dict:
        """查詢指定股票的持倉"""
        positions = self.ctx.stock_positions()
        
        for channel in positions.channels:
            for pos in channel.positions:
                if pos.symbol == symbol:
                    info = {
                        'symbol': pos.symbol,
                        'quantity': int(pos.quantity),
                        'available_quantity': int(pos.available_quantity),
                        'cost_price': float(pos.cost_price),
                        'market_value': float(pos.market),
                    }
                    logger.info(
                        f"持倉 {symbol} | 數量: {info['quantity']} 股 | "
                        f"成本: ${info['cost_price']:.2f} | "
                        f"市值: ${info['market_value']:,.2f}"
                    )
                    return info
        
        logger.info(f"當前無 {symbol} 持倉")
        return {'symbol': symbol, 'quantity': 0, 'available_quantity': 0}
    
    def buy(self, symbol="SPY.US", quantity=1) -> str:
        """
        市價買入
        
        Parameters:
            symbol: 股票代碼
            quantity: 買入股數
            
        Returns:
            訂單 ID
        """
        logger.info(f"{'[模擬]' if self.simulate else '[實盤]'} "
                     f"買入 {symbol} x {quantity} 股 (市價單)")
        
        if self.simulate:
            logger.info("模擬模式：跳過實際下單")
            return "SIM_ORDER_BUY_001"
        
        response = self.ctx.submit_order(
            side=OrderSide.Buy,
            symbol=symbol,
            order_type=OrderType.MO,
            submitted_quantity=Decimal(str(quantity)),
            time_in_force=TimeInForceType.Day,
            outside_rth=OutsideRTH.AnyTime,
            remark="Zero2Quant EMA Strategy BUY",
        )
        
        order_id = response.order_id
        logger.info(f"✅ 買入訂單已提交 | 訂單ID: {order_id}")
        return order_id
    
    def sell(self, symbol="SPY.US", quantity=None) -> str:
        """
        市價賣出
        
        Parameters:
            symbol: 股票代碼
            quantity: 賣出股數（None = 全部賣出）
            
        Returns:
            訂單 ID
        """
        # 如果未指定數量，查詢持倉全部賣出
        if quantity is None:
            pos = self.get_positions(symbol)
            quantity = pos.get('available_quantity', 0)
            if quantity == 0:
                logger.warning(f"無可用持倉可賣出: {symbol}")
                return ""
        
        logger.info(f"{'[模擬]' if self.simulate else '[實盤]'} "
                     f"賣出 {symbol} x {quantity} 股 (市價單)")
        
        if self.simulate:
            logger.info("模擬模式：跳過實際下單")
            return "SIM_ORDER_SELL_001"
        
        response = self.ctx.submit_order(
            side=OrderSide.Sell,
            symbol=symbol,
            order_type=OrderType.MO,
            submitted_quantity=Decimal(str(quantity)),
            time_in_force=TimeInForceType.Day,
            outside_rth=OutsideRTH.AnyTime,
            remark="Zero2Quant EMA Strategy SELL",
        )
        
        order_id = response.order_id
        logger.info(f"✅ 賣出訂單已提交 | 訂單ID: {order_id}")
        return order_id
    
    def get_order_status(self, order_id: str) -> dict:
        """查詢訂單狀態"""
        order = self.ctx.order_detail(order_id=order_id)
        
        info = {
            'order_id': order.order_id,
            'status': str(order.status),
            'symbol': order.symbol,
            'side': str(order.side),
            'quantity': int(order.quantity),
            'executed_quantity': int(order.executed_quantity),
            'price': float(order.price) if order.price else None,
            'executed_price': float(order.executed_price) if order.executed_price else None,
        }
        
        logger.info(f"訂單 {order_id} | 狀態: {info['status']} | "
                     f"已成交: {info['executed_quantity']}/{info['quantity']}")
        return info
    
    def cancel_order(self, order_id: str):
        """撤銷訂單"""
        logger.info(f"撤銷訂單: {order_id}")
        
        if self.simulate:
            logger.info("模擬模式：跳過實際撤單")
            return
        
        self.ctx.cancel_order(order_id=order_id)
        logger.info(f"✅ 訂單已撤銷: {order_id}")


# ===== 使用示例 =====
if __name__ == "__main__":
    # 以模擬模式運行
    executor = TradeExecutor(simulate=True)
    
    # 查詢賬戶餘額
    balance = executor.get_account_balance()
    
    # 查詢持倉
    position = executor.get_positions("SPY.US")
    
    # 模擬買入 10 股
    order_id = executor.buy("SPY.US", quantity=10)
    
    # 模擬全倉賣出
    order_id = executor.sell("SPY.US")

⚠️ 重要提醒：

首次運行時務必使用 simulate=True 模擬模式！確認邏輯無誤後，再切換到實盤模式（simulate=False）。

6. 第四步：用 AI 整合完整交易系統

Prompt #4：整合策略與交易執行

📝 複製這段 Prompt

我現在有兩個模塊： 1. EMACrossoverStrategy：計算交易信號 2. TradeExecutor：執行交易請幫我構建一個完整的交易系統主程序，將兩者整合。要求： 1. 定時執行（每天美股收盤後運行一次） 2. 流程：獲取最新數據 → 計算信號 → 判斷是否需要交易 → 執行下單 3. 加入倉位管理： - 無持倉 + 買入信號 → 買入（用可用資金的 90% 計算股數） - 有持倉 + 賣出信號 → 全倉賣出 - 其他情況 → 不操作 4. 完整的日誌記錄 5. 異常處理（網絡錯誤、API 錯誤等）請提供完整代碼。

AI 生成的代碼

from longport.openapi import QuoteContext, Config, Period, AdjustType
import pandas as pd
import logging
import schedule
import time
from datetime import datetime

# 導入自定義模塊
# from strategy import EMACrossoverStrategy
# from executor import TradeExecutor

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler('trading.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)


class SPYTradingSystem:
    """
    SPY 自動交易系統
    
    整合 Longport 行情 + 策略計算 + 交易執行
    """
    
    def __init__(self, simulate=True):
        self.simulate = simulate
        self.config = Config.from_env()
        self.symbol = "SPY.US"
        
        # 初始化模塊
        self.strategy = EMACrossoverStrategy(
            fast_period=20,
            slow_period=60,
            trend_period=200
        )
        self.executor = TradeExecutor(simulate=simulate)
        
        logger.info(f"交易系統初始化完成 | 標的: {self.symbol}")
    
    def fetch_data(self) -> pd.DataFrame:
        """獲取最新行情數據"""
        logger.info("正在獲取行情數據...")
        
        ctx = QuoteContext(self.config)
        
        candlesticks = ctx.candlesticks(
            symbol=self.symbol,
            period=Period.Day,
            count=300,  # 獲取足夠的數據計算 200 日均線
            adjust_type=AdjustType.ForwardAdjust
        )
        
        records = [{
            'date': c.timestamp.strftime('%Y-%m-%d'),
            'open': float(c.open),
            'high': float(c.high),
            'low': float(c.low),
            'close': float(c.close),
            'volume': int(c.volume),
        } for c in candlesticks]
        
        df = pd.DataFrame(records)
        df['date'] = pd.to_datetime(df['date'])
        df = df.sort_values('date').reset_index(drop=True)
        
        logger.info(f"數據獲取完成 | {len(df)} 條 K 線 | "
                     f"最新日期: {df['date'].iloc[-1].strftime('%Y-%m-%d')}")
        return df
    
    def calculate_position_size(self, price: float) -> int:
        """計算買入股數（使用 90% 可用資金）"""
        balance = self.executor.get_account_balance()
        available = balance.get('available', 0)
        
        # 使用 90% 資金
        buy_amount = available * 0.9
        quantity = int(buy_amount / price)
        
        logger.info(f"倉位計算 | 可用資金: ${available:,.2f} | "
                     f"目標金額: ${buy_amount:,.2f} | "
                     f"買入股數: {quantity}")
        return quantity
    
    def run_once(self):
        """執行一次交易邏輯"""
        logger.info("=" * 60)
        logger.info(f"開始執行交易邏輯 | {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        logger.info("=" * 60)
        
        try:
            # Step 1: 獲取數據
            df = self.fetch_data()
            
            # Step 2: 計算信號
            signals_df = self.strategy.calculate_signals(df)
            signal_info = self.strategy.get_latest_signal()
            self.strategy.print_signal_summary()
            
            signal = signal_info['signal']
            latest_price = signal_info['close']
            
            # Step 3: 查詢持倉
            position = self.executor.get_positions(self.symbol)
            has_position = position.get('quantity', 0) > 0
            
            # Step 4: 交易決策
            if signal == 'BUY' and not has_position:
                logger.info("📈 策略信號: 買入 | 當前無持倉 → 執行買入")
                quantity = self.calculate_position_size(latest_price)
                if quantity > 0:
                    order_id = self.executor.buy(self.symbol, quantity)
                    logger.info(f"買入訂單已提交 | 股數: {quantity} | 預計金額: ${quantity * latest_price:,.2f}")
                else:
                    logger.warning("資金不足，無法買入")
            
            elif signal == 'SELL' and has_position:
                logger.info("📉 策略信號: 賣出 | 當前有持倉 → 執行賣出")
                order_id = self.executor.sell(self.symbol)
                logger.info(f"賣出訂單已提交 | 清倉 {position['quantity']} 股")
            
            else:
                if has_position:
                    logger.info(f"⏸️ 當前持倉 {position['quantity']} 股 | 信號: {signal} | 不操作")
                else:
                    logger.info(f"⏸️ 當前無持倉 | 信號: {signal} | 等待入場機會")
            
            logger.info("交易邏輯執行完成\n")
            
        except Exception as e:
            logger.error(f"❌ 交易執行異常: {e}", exc_info=True)
    
    def run_scheduled(self, run_time="16:30"):
        """
        定時運行（每天指定時間執行一次）
        
        Parameters:
            run_time: 執行時間（美東時間 16:30 = 收盤後 30 分鐘）
        """
        logger.info(f"定時任務已啟動 | 每天 {run_time} 執行")
        
        schedule.every().monday.at(run_time).do(self.run_once)
        schedule.every().tuesday.at(run_time).do(self.run_once)
        schedule.every().wednesday.at(run_time).do(self.run_once)
        schedule.every().thursday.at(run_time).do(self.run_once)
        schedule.every().friday.at(run_time).do(self.run_once)
        
        while True:
            schedule.run_pending()
            time.sleep(60)


# ===== 主程序入口 =====
if __name__ == "__main__":
    # 初始化交易系統（模擬模式）
    system = SPYTradingSystem(simulate=True)
    
    # 手動執行一次（測試用）
    system.run_once()
    
    # 正式運行：取消下面的註釋
    # system.run_scheduled(run_time="16:30")

運行結果示例

2026-02-16 16:30:01 [INFO] ============================================================
2026-02-16 16:30:01 [INFO] 開始執行交易邏輯 | 2026-02-16 16:30:01
2026-02-16 16:30:01 [INFO] ============================================================
2026-02-16 16:30:02 [INFO] 正在獲取行情數據...
2026-02-16 16:30:03 [INFO] 數據獲取完成 | 300 條 K 線 | 最新日期: 2026-02-14
2026-02-16 16:30:03 [INFO] ==================================================
2026-02-16 16:30:03 [INFO] 📊 策略信號摘要
2026-02-16 16:30:03 [INFO] ==================================================
2026-02-16 16:30:03 [INFO] 日期:       2026-02-14
2026-02-16 16:30:03 [INFO] 收盤價:     $612.35
2026-02-16 16:30:03 [INFO] EMA20:      $608.72
2026-02-16 16:30:03 [INFO] EMA60:      $601.15
2026-02-16 16:30:03 [INFO] SMA200:     $575.48
2026-02-16 16:30:03 [INFO] 趨勢過濾:   ✅ 趨勢向上
2026-02-16 16:30:03 [INFO] ==================================================
2026-02-16 16:30:03 [INFO] 當前信號:   ⏸️ 持有
2026-02-16 16:30:03 [INFO] ==================================================
2026-02-16 16:30:03 [INFO] 當前無 SPY.US 持倉
2026-02-16 16:30:03 [INFO] ⏸️ 當前無持倉 | 信號: HOLD | 等待入場機會
2026-02-16 16:30:03 [INFO] 交易邏輯執行完成

7. 第五步：用 AI 添加風控與監控

Prompt #5：添加風控模塊

📝 複製這段 Prompt

最後，請幫我添加風險控制和監控模塊。風控要求： 1. 單筆最大虧損限制：持倉虧損超過 5% 自動止損 2. 單日最大交易次數限制：每天最多交易 2 次 3. 最大持倉比例：不超過賬戶總資產的 90% 4. 交易時段檢查：只在美股交易時段（9:30-16:00 ET）內下單監控要求： 1. 每次交易後記錄到 CSV 日誌 2. 打印當天盈虧統計請在現有代碼基礎上添加。

AI 生成的風控代碼

import csv
import os
from datetime import datetime, time
import pytz

class RiskManager:
    """
    風險管理器
    
    - 止損控制
    - 交易頻率限制
    - 倉位比例控制
    - 交易時段檢查
    """
    
    def __init__(self, 
                 max_loss_pct=5.0,
                 max_daily_trades=2,
                 max_position_pct=90.0,
                 trade_log_file='trade_log.csv'):
        
        self.max_loss_pct = max_loss_pct
        self.max_daily_trades = max_daily_trades
        self.max_position_pct = max_position_pct
        self.trade_log_file = trade_log_file
        self.daily_trade_count = 0
        self.last_trade_date = None
        
        # 初始化交易日誌
        self._init_trade_log()
        
        logger.info(f"風控模塊初始化 | 止損: {max_loss_pct}% | "
                     f"日交易上限: {max_daily_trades} | "
                     f"最大倉位: {max_position_pct}%")
    
    def _init_trade_log(self):
        """初始化交易日誌 CSV"""
        if not os.path.exists(self.trade_log_file):
            with open(self.trade_log_file, 'w', newline='') as f:
                writer = csv.writer(f)
                writer.writerow([
                    'timestamp', 'symbol', 'side', 'quantity',
                    'price', 'amount', 'signal', 'pnl_pct'
                ])
    
    def is_trading_hours(self) -> bool:
        """檢查是否在美股交易時段"""
        et = pytz.timezone('US/Eastern')
        now = datetime.now(et)
        
        # 美股交易時間: 9:30 - 16:00 ET
        market_open = time(9, 30)
        market_close = time(16, 0)
        
        # 檢查是否為工作日（0=Monday, 4=Friday）
        if now.weekday() > 4:
            logger.warning("⚠️ 非交易日（週末）")
            return False
        
        current_time = now.time()
        if current_time < market_open or current_time > market_close:
            logger.warning(f"⚠️ 非交易時段 | 當前: {current_time} ET | "
                          f"交易時段: {market_open}-{market_close}")
            return False
        
        return True
    
    def check_daily_limit(self) -> bool:
        """檢查今日交易次數限制"""
        today = datetime.now().strftime('%Y-%m-%d')
        
        if self.last_trade_date != today:
            self.daily_trade_count = 0
            self.last_trade_date = today
        
        if self.daily_trade_count >= self.max_daily_trades:
            logger.warning(f"⚠️ 已達今日交易上限: {self.daily_trade_count}/{self.max_daily_trades}")
            return False
        
        return True
    
    def check_stop_loss(self, cost_price: float, current_price: float) -> bool:
        """
        檢查是否需要止損
        
        Returns:
            True = 需要止損，False = 不需要
        """
        if cost_price <= 0:
            return False
        
        loss_pct = (current_price - cost_price) / cost_price * 100
        
        if loss_pct < -self.max_loss_pct:
            logger.warning(f"🚨 觸發止損! | 成本: ${cost_price:.2f} | "
                          f"現價: ${current_price:.2f} | "
                          f"虧損: {loss_pct:.2f}%（超過 -{self.max_loss_pct}%）")
            return True
        
        return False
    
    def check_position_limit(self, buy_amount: float, total_assets: float) -> bool:
        """檢查倉位比例限制"""
        if total_assets <= 0:
            return False
        
        position_pct = buy_amount / total_assets * 100
        
        if position_pct > self.max_position_pct:
            logger.warning(f"⚠️ 倉位比例超限: {position_pct:.1f}% > {self.max_position_pct}%")
            return False
        
        return True
    
    def log_trade(self, symbol, side, quantity, price, signal, pnl_pct=0):
        """記錄交易到 CSV"""
        with open(self.trade_log_file, 'a', newline='') as f:
            writer = csv.writer(f)
            writer.writerow([
                datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                symbol, side, quantity,
                f"{price:.2f}",
                f"{quantity * price:.2f}",
                signal,
                f"{pnl_pct:.2f}"
            ])
        
        self.daily_trade_count += 1
        logger.info(f"📝 交易已記錄 | {side} {symbol} x {quantity} @ ${price:.2f}")
    
    def can_trade(self, side: str, cost_price=0, current_price=0,
                  buy_amount=0, total_assets=0) -> bool:
        """
        綜合風控檢查
        
        Returns:
            True = 允許交易
        """
        # 1. 交易時段檢查
        if not self.is_trading_hours():
            return False
        
        # 2. 日交易次數檢查
        if not self.check_daily_limit():
            return False
        
        # 3. 買入時檢查倉位比例
        if side == 'BUY' and total_assets > 0:
            if not self.check_position_limit(buy_amount, total_assets):
                return False
        
        logger.info("✅ 風控檢查通過")
        return True

✅ 風控體系亮點：

自動止損線保護 — 虧損超過 5% 自動清倉保命
交易頻率限制 — 避免頻繁交易產生過多手續費
倉位控制 — 永遠留 10% 現金作為安全墊
時段檢查 — 只在開市期間下單，避免盤後流動性風險

8. 實盤交易注意事項與技巧總結

8.1 從模擬到實盤的過渡步驟

模擬驗證：使用 simulate=True 運行至少 2 週，確認邏輯正確
小倉位測試：先用 1 股 SPY 實盤測試下單流程
監控日誌：檢查 trade_log.csv 確認每筆交易是否合理
逐步加倉：確認無誤後，逐步增加倉位比例

8.2 與 AI 交互構建實盤系統的 Prompt 技巧

技巧	說明	示例
指定 SDK 版本	告訴 AI 你用的是哪個 SDK	「使用 longport Python SDK 最新版」
強調安全性	提醒 AI 加入保護機制	「必須支持模擬模式」
給出 API 文檔	如果 AI 不了解 SDK，附上文檔鏈接	「參考 open.longportapp.com 文檔」
分模塊推進	不要一次性要求全部功能	「先實現行情獲取，再實現下單」
強調錯誤處理	實盤系統必須穩健	「加入 try-except 和日誌記錄」

8.3 完整系統架構回顧

┌─────────────────────────────────────────┐
│           SPY Trading System            │
├─────────────────────────────────────────┤
│                                         │
│   ┌─────────────┐   ┌───────────────┐   │
│   │  Longport    │   │   Strategy    │   │
│   │  Quote API   │──▶│   Engine      │   │
│   │  (行情數據)   │   │  (信號計算)    │   │
│   └─────────────┘   └──────┬────────┘   │
│                            │             │
│                     ┌──────▼────────┐    │
│                     │  Risk Manager │    │
│                     │  (風控檢查)    │    │
│                     └──────┬────────┘    │
│                            │             │
│                     ┌──────▼────────┐    │
│                     │  Longport     │    │
│                     │  Trade API    │    │
│                     │  (下單執行)    │    │
│                     └──────┬────────┘    │
│                            │             │
│                     ┌──────▼────────┐    │
│                     │  Trade Log    │    │
│                     │  (交易日誌)    │    │
│                     └───────────────┘    │
│                                         │
└─────────────────────────────────────────┘

8.4 後續擴展方向

多標的支持：同時監控 QQQ、IWM 等 ETF
WebSocket 實時推送：使用 Longport 推送 API 替代定時輪詢
通知系統：交易信號觸發時發送 Telegram / Email 通知
數據庫存儲：用 SQLite / PostgreSQL 替代 CSV 日誌
Web 面板：用 Flask / Streamlit 構建監控儀表盤

🎉 總結：

通過 5 輪 AI 對話，我們完成了一個完整的 Longport SDK 實盤交易系統：行情獲取 → 策略信號 → 下單執行 → 風控監控。從回測到實盤，關鍵在於：模塊化設計 + 嚴格風控 + 充分測試。

想要更系統地學習量化交易和實盤對接？在我的課程中，你將學會更多策略設計和實盤技巧，包含完整的 Longport SDK 實戰項目。

如何和 AI 交互構建對接 Longport SDK 的 SPY 實盤交易策略

目錄

1. 為什麼選擇 Longport SDK？

2. 環境準備與 API 配置

2.1 安裝 Longport SDK

2.2 獲取 API 密鑰

2.3 配置環境變量

3. 第一步：用 AI 獲取 SPY 實時行情數據

Prompt #1：通過 Longport SDK 獲取 SPY 行情

AI 生成的代碼

4. 第二步：用 AI 構建均線突破策略邏輯

Prompt #2：實現策略信號計算

AI 生成的代碼

5. 第三步：用 AI 實現實盤下單模塊

Prompt #3：Longport 下單與持倉管理

AI 生成的代碼

6. 第四步：用 AI 整合完整交易系統

Prompt #4：整合策略與交易執行

AI 生成的代碼

運行結果示例

7. 第五步：用 AI 添加風控與監控

Prompt #5：添加風控模塊

AI 生成的風控代碼

8. 實盤交易注意事項與技巧總結

8.1 從模擬到實盤的過渡步驟

8.2 與 AI 交互構建實盤系統的 Prompt 技巧

8.3 完整系統架構回顧

8.4 後續擴展方向

江景哲

相關文章

想要更系統地學習量化交易？