你的第一个 AI 交易机器人：EMA 交叉 + Alpaca 模拟交易

python -m venv venv
source venv/bin/activate  # Windows 用：venv\Scripts\activate
pip install alpaca-py pandas numpy python-dotenv yfinance

import os
from datetime import datetime, timedelta
from dotenv import load_dotenv
from alpaca.data.historical import StockHistoricalDataClient
from alpaca.data.requests import StockBarsRequest
from alpaca.data.timeframe import TimeFrame
import pandas as pd
 
load_dotenv()
 
client = StockHistoricalDataClient(
    api_key=os.getenv("ALPACA_API_KEY"),
    secret_key=os.getenv("ALPACA_SECRET_KEY"),
    raw_data=False,
)
 
request = StockBarsRequest(
    symbol_or_symbols="SPY",
    timeframe=TimeFrame.Hour,
    start=datetime.now() - timedelta(days=90),
    end=datetime.now(),
    feed="iex",
)
 
bars = client.get_stock_bars(request)
df = bars.df.reset_index()
df = df.rename(columns={"timestamp": "timestamp", "open": "open", "high": "high", "low": "low", "close": "close", "volume": "volume"})
print(df.tail())

import pandas as pd
 
def add_ema_signals(df: pd.DataFrame, fast: int = 12, slow: int = 26) -> pd.DataFrame:
    df = df.copy()
    df["ema_fast"] = df["close"].ewm(span=fast, adjust=False).mean()
    df["ema_slow"] = df["close"].ewm(span=slow, adjust=False).mean()
    df["signal"] = 0
    df.loc[df["ema_fast"] > df["ema_slow"], "signal"] = 1
    df.loc[df["ema_fast"] < df["ema_slow"], "signal"] = -1
    df["crossover"] = df["signal"].diff()
    return df

import pandas as pd
from data import get_bars
from strategy import add_ema_signals
 
INITIAL_CAPITAL = 10000
COMMISSION_PER_TRADE = 1.0
 
def backtest(df: pd.DataFrame, fast: int = 12, slow: int = 26):
    df = add_ema_signals(df, fast, slow)
    position = 0
    cash = INITIAL_CAPITAL
    shares = 0
    trades = []
 
    for _, row in df.iterrows():
        price = row["close"]
        crossover = row["crossover"]
 
        if crossover == 2 and position <= 0:
            # 金叉：买入
            shares = (cash - COMMISSION_PER_TRADE) // price
            if shares > 0:
                cash -= shares * price + COMMISSION_PER_TRADE
                position = 1
                trades.append({"type": "buy", "price": price, "timestamp": row["timestamp"]})
        elif crossover == -2 and position >= 0:
            # 死叉：卖出
            if shares > 0:
                cash += shares * price - COMMISSION_PER_TRADE
                trades.append({"type": "sell", "price": price, "timestamp": row["timestamp"]})
            shares = 0
            position = -1
 
    final_value = cash + shares * df.iloc[-1]["close"]
    return {
        "final_value": final_value,
        "return_pct": (final_value - INITIAL_CAPITAL) / INITIAL_CAPITAL * 100,
        "trades": len(trades),
        "trade_list": trades,
    }
 
if __name__ == "__main__":
    df = get_bars("SPY", days=180)
    result = backtest(df, fast=12, slow=26)
    print(f"最终价值: ${result['final_value']:.2f}")
    print(f"收益率: {result['return_pct']:.2f}%")
    print(f"交易次数: {result['trades']}")

def max_drawdown(equity_curve: pd.Series) -> float:
    rolling_max = equity_curve.cummax()
    drawdown = (equity_curve - rolling_max) / rolling_max
    return drawdown.min()

import os
import time
from datetime import datetime
from dotenv import load_dotenv
from alpaca.trading.client import TradingClient
from alpaca.trading.requests import MarketOrderRequest
from alpaca.trading.enums import OrderSide, TimeInForce
from data import get_bars
from strategy import add_ema_signals
 
load_dotenv()
 
SYMBOL = "SPY"
QTY = 10
FAST = 12
SLOW = 26
 
trading_client = TradingClient(
    api_key=os.getenv("ALPACA_API_KEY"),
    secret_key=os.getenv("ALPACA_SECRET_KEY"),
    paper=True,
)
 
def get_latest_signal():
    df = get_bars(SYMBOL, days=5)
    df = add_ema_signals(df, fast=FAST, slow=SLOW)
    return df.iloc[-1]["signal"], df.iloc[-2]["signal"]
 
def place_order(side: OrderSide, qty: int):
    order = MarketOrderRequest(
        symbol=SYMBOL,
        qty=qty,
        side=side,
        time_in_force=TimeInForce.DAY,
    )
    return trading_client.submit_order(order)
 
if __name__ == "__main__":
    current, previous = get_latest_signal()
    if current == 1 and previous != 1:
        print(f"{datetime.now()}: 金叉。买入 {QTY} 股 {SYMBOL}。")
        place_order(OrderSide.BUY, QTY)
    elif current == -1 and previous != -1:
        print(f"{datetime.now()}: 死叉。卖出 {QTY} 股 {SYMBOL}。")
        place_order(OrderSide.SELL, QTY)
    else:
        print(f"{datetime.now()}: 无交叉。不操作。")

import logging
logging.basicConfig(
    filename="bot.log",
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)