Alpaca SMA交易算法问题排查：代码警告与潜在Bug分析

问题概述 #

这段50日均线交易策略代码存在4个核心问题，同时触发pandas版本兼容警告：

• FutureWarning：直接用[-1]访问Series末尾元素的方式将被弃用
• 持仓检查逻辑错误：未遍历完所有持仓就提前返回0，导致漏判其他持仓
• 交易逻辑违规：无持仓时直接发起做空订单不符合常规交易规则
• 性能低下：每次执行交易函数都重复下载全量历史数据

逐个问题修复 #

1. 解决FutureWarning #

将所有按位置访问Series末尾元素的代码historical_data['SMA_50'][-1]替换为pandas官方推荐的historical_data['SMA_50'].iloc[-1]，避免版本兼容问题。

2. 修复持仓检查函数 #

原函数在遍历第一个持仓时，若不是目标标的就直接返回0，正确逻辑是遍历完所有持仓后，未找到目标标的才返回0：

def check_positions(symbol):
    positions = api.list_positions()
    for position in positions:
        if position.symbol == symbol:
            return int(position.qty)
    return 0  # 遍历完所有持仓后再返回0

3. 修正做空逻辑 #

常规策略中，无持仓时不应直接做空，仅当持有多头仓位时，价格跌破均线才卖出平仓。若需做空需单独添加权限判断，此处调整为仅平仓已有多头：

elif current_price < historical_data['SMA_50'].iloc[-1]:
    current_qty = check_positions(symbol)
    if current_qty > 0:
        api.submit_order(symbol=symbol,
                         qty=current_qty,
                         side='sell',
                         type='market',
                         time_in_force='gtc')
        print(f"Sell order placed for {symbol} {current_qty}")
    else:
        print("No long position to sell, skip short order")

4. 优化数据获取效率 #

将历史数据初始化和SMA计算移至交易函数外部，每次交易仅获取增量数据更新，避免重复下载全量数据：

# 初始化全局数据
historical_data = yf.download(symbol, start=start_date, end=pd.Timestamp.today().strftime('%Y-%m-%d'))
historical_data['SMA_50'] = historical_data['Close'].rolling(window=50).mean()

# 更新增量数据
def update_data():
    global historical_data
    latest_start = historical_data.index[-1] + pd.Timedelta(days=1)
    latest_data = yf.download(symbol, start=latest_start.strftime('%Y-%m-%d'), end=pd.Timestamp.today().strftime('%Y-%m-%d'))
    if not latest_data.empty:
        historical_data = pd.concat([historical_data, latest_data])
        historical_data['SMA_50'] = historical_data['Close'].rolling(window=50).mean()

完整修复后的代码 #

from alpaca_trade_api import REST
import time
import pandas as pd
import yfinance as yf

# Alpaca API配置
api_key = 'PKXE9H6DCGDZ8LU60MAV'
api_secret = 'Zym9PVK7RqgS6mQy0PVhfT4jAwxXCsktQRg25cLT'
base_url = 'https://paper-api.alpaca.markets'

api = REST(api_key, api_secret, base_url)

# 策略参数
symbol = 'AAPL'
qty = 1
start_date = '2015-01-01'

# 初始化历史数据和50日均线
def init_data():
    global historical_data
    end_date = pd.Timestamp.today().strftime('%Y-%m-%d')
    historical_data = yf.download(symbol, start=start_date, end=end_date)
    historical_data['SMA_50'] = historical_data['Close'].rolling(window=50).mean()

# 更新增量数据（避免重复下载全量）
def update_data():
    global historical_data
    if historical_data.empty:
        init_data()
        return
    latest_start = historical_data.index[-1] + pd.Timedelta(days=1)
    latest_end = pd.Timestamp.today()
    latest_data = yf.download(symbol, start=latest_start.strftime('%Y-%m-%d'), end=latest_end.strftime('%Y-%m-%d'))
    if not latest_data.empty:
        historical_data = pd.concat([historical_data, latest_data])
        historical_data['SMA_50'] = historical_data['Close'].rolling(window=50).mean()

# 检查当前持仓数量
def check_positions(symbol):
    positions = api.list_positions()
    for position in positions:
        if position.symbol == symbol:
            return int(position.qty)
    return 0

# 交易执行逻辑
def trade(symbol, qty):
    update_data()
    # 检查数据量是否足够计算SMA50
    if len(historical_data) < 50:
        print("Insufficient data to calculate SMA 50")
        return
    
    current_price = api.get_latest_trade(symbol).price
    latest_sma = historical_data['SMA_50'].iloc[-1]
    current_qty = check_positions(symbol)

    if current_price > latest_sma:
        if current_qty == 0:
            api.submit_order(symbol=symbol,
                             qty=qty,
                             side='buy',
                             type='market',
                             time_in_force='gtc')
            print(f"Buy order placed for {symbol} {qty} shares at ${current_price:.2f}")
        elif current_qty == qty:
            print(f"Holding {symbol} {qty} shares")
        else:
            print(f"Current position {current_qty} shares, no action needed")
    elif current_price < latest_sma:
        if current_qty > 0:
            api.submit_order(symbol=symbol,
                             qty=current_qty,
                             side='sell',
                             type='market',
                             time_in_force='gtc')
            print(f"Sell order placed for {symbol} {current_qty} shares at ${current_price:.2f}")
        else:
            print("No long position to sell, skip short order")
    else:
        print(f"{symbol} price equals SMA 50, no action taken")

# 初始化数据
init_data()

# 每日执行交易
while True:
    trade(symbol, qty)
    time.sleep(86400)

额外优化点 #

• 移除重复的symbol定义
• 修复time_in_force参数中的多余空格
• 添加数据量不足时的判断逻辑
• 格式化输出信息，增加实时价格显示
• 优化数据更新逻辑，仅获取增量数据

内容的提问来源于stack exchange，提问作者Wyatt Gill