使用Python实现高效自动化交易：从数据获取到策略执行的全面指南

import yfinance as yf
​
# 获取股票数据
stock_data = yf.download('AAPL', start='2023-01-01', end='2024-01-01')
print(stock_data.head())

import pandas as pd
​
# 计算移动平均线
stock_data['MA_20'] = stock_data['Close'].rolling(window=20).mean()
stock_data['MA_50'] = stock_data['Close'].rolling(window=50).mean()
​
# 生成交易信号
stock_data['Signal'] = 0
stock_data['Signal'][20:] = np.where(stock_data['MA_20'][20:] > stock_data['MA_50'][20:], 1, 0)
​
# 计算持仓信号
stock_data['Position'] = stock_data['Signal'].diff()
​
print(stock_data.head())

import alpaca_trade_api as tradeapi
​
api = tradeapi.REST('<API Key>', '<API Secret>', base_url='https://paper-api.alpaca.markets')
​
# 获取账户信息
account = api.get_account()
​
# 下单
if stock_data['Position'].iloc[-1] == 1 and account.buying_power > 0:
    api.submit_order(
        symbol='AAPL',
        qty=10,
        side='buy',
        type='market',
        time_in_force='gtc'
    )
elif stock_data['Position'].iloc[-1] == -1:
    api.submit_order(
        symbol='AAPL',
        qty=10,
        side='sell',
        type='market',
        time_in_force='gtc'
    )

# 计算投资组合价值
portfolio_value = stock_data['Close'] * 10
​
# 计算日收益率
daily_returns = portfolio_value.pct_change()
​
# 计算投资组合风险
portfolio_std_dev = daily_returns.std()
portfolio_annualized_volatility = portfolio_std_dev * np.sqrt(252)
​
print("投资组合年化波动率:", portfolio_annualized_volatility)

import backtrader as bt
​
# 创建策略
class MovingAverageStrategy(bt.Strategy):
    params = (('sma_period', 20), ('lma_period', 50))
​
    def __init__(self):
        self.sma = bt.indicators.SimpleMovingAverage(self.data, period=self.params.sma_period)
        self.lma = bt.indicators.SimpleMovingAverage(self.data, period=self.params.lma_period)
​
    def next(self):
        if self.sma > self.lma:
            self.buy()
        elif self.sma < self.lma:
            self.sell()
​
# 创建回测引擎
cerebro = bt.Cerebro()
​
# 添加数据
data = bt.feeds.PandasData(dataname=stock_data)
​
cerebro.adddata(data)
​
# 添加策略
cerebro.addstrategy(MovingAverageStrategy)
​
# 设置初始资金
cerebro.broker.set_cash(100000)
​
# 运行回测
cerebro.run()
​
# 获取回测结果
cerebro.plot()

import websocket
import json
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
​
# WebSocket回调函数
def on_message(ws, message):
    data = json.loads(message)
    price = data['price']
    if price > 100:
        send_email("AAPL价格超过100美元", f"AAPL价格已经超过100美元，当前价格为{price}")
​
# 发送邮件函数
def send_email(subject, message):
    sender_email = "your_email@gmail.com"
    receiver_email = "recipient_email@gmail.com"
    password = "your_email_password"
​
    msg = MIMEMultipart()
    msg['From'] = sender_email
    msg['To'] = receiver_email
    msg['Subject'] = subject
​
    msg.attach(MIMEText(message, 'plain'))
​
    server = smtplib.SMTP('smtp.gmail.com', 587)
    server.starttls()
    server.login(sender_email, password)
    text = msg.as_string()
    server.sendmail(sender_email, receiver_email, text)
    server.quit()
​
# 连接WebSocket
ws = websocket.WebSocketApp("wss://ws.finnhub.io?token=your_token", on_message=on_message)
ws.run_forever()

import matplotlib.pyplot as plt
​
# 绘制资产曲线
plt.plot(portfolio_value.index, portfolio_value.values)
plt.title('Portfolio Value')
plt.xlabel('Date')
plt.ylabel('Value')
plt.show()
​
# 绘制交易心态分析
plt.hist(daily_returns, bins=20)
plt.title('Daily Returns Distribution')
plt.xlabel('Daily Returns')
plt.ylabel('Frequency')
plt.show()