企业微信

如何用Python批量获取股票数据并分析策略

数据获取与预处理

批量获取股票数据是量化交易的基础工作。Python生态提供了多种工具，其中yfinance库可以免费获取全球多个市场的股票和期货数据。安装命令：pip install yfinance pandas numpy matplotlib。以下代码演示如何批量下载多只股票和期货合约的日线数据，并保存为CSV文件。


import yfinance as yf

import pandas as pd

# 定义股票和期货代码列表

tickers = ['AAPL', 'MSFT', 'GOOGL', 'CL=F', 'GC=F']  # 股票与期货混合

# 批量下载数据

data = yf.download(tickers, start='2020-01-01', end='2024-01-01', group_by='ticker')

# 保存每个标的的数据到独立文件

for ticker in tickers:

    df = data[ticker].copy()

    df.to_csv(f'{ticker}_data.csv')

    print(f'{ticker} 数据已保存，共 {len(df)} 条记录')

下载后的数据包含开盘价、最高价、最低价、收盘价、成交量等字段。预处理阶段需要处理缺失值、调整股价（复权）、计算收益率。对于期货数据，还需注意主力合约的拼接问题。

如何用Python批量获取股票数据并分析策略

策略构建与回测框架

量化策略的核心是定义买卖信号。以双均线交叉策略为例，当短期均线上穿长期均线时买入，下穿时卖出。以下代码构建一个简易回测框架，支持批量测试多个标的。


import pandas as pd

import numpy as np

def dual_moving_average_strategy(df, short_window=20, long_window=50):

    df = df.copy()

    df['short_ma'] = df['Close'].rolling(window=short_window).mean()

    df['long_ma'] = df['Close'].rolling(window=long_window).mean()

    df['signal'] = 0

    df.loc[df['short_ma'] > df['long_ma'], 'signal'] = 1

    df['position'] = df['signal'].diff()

    return df

def backtest(df, initial_capital=100000):

    df = df.dropna()

    df['returns'] = df['Close'].pct_change()

    df['strategy_returns'] = df['signal'].shift(1) * df['returns']

    df['equity'] = initial_capital * (1 + df['strategy_returns']).cumprod()

    total_return = df['equity'].iloc[-1] / initial_capital - 1

    annual_return = (1 + total_return) ** (252 / len(df)) - 1

    sharpe = df['strategy_returns'].mean() / df['strategy_returns'].std() * np.sqrt(252)

    max_drawdown = (df['equity'] / df['equity'].cummax() - 1).min()

    return {'total_return': total_return, 'annual_return': annual_return,

            'sharpe': sharpe, 'max_drawdown': max_drawdown}

遍历所有标的，输出绩效指标。期货品种如原油（CL=F）和黄金（GC=F）同样适用该框架，但需注意期货的杠杆和保证金机制，回测时应调整仓位管理。


results = {}

for ticker in tickers:

    df = pd.read_csv(f'{ticker}_data.csv', index_col=0, parse_dates=True)

    df = dual_moving_average_strategy(df)

    results[ticker] = backtest(df)

    print(f'{ticker}: {results[ticker]}')

绩效评估与风险控制

回测结果需要多维度评估。年化收益反映策略盈利能力，夏普比率衡量单位风险超额收益，最大回撤揭示最坏情况下的亏损幅度。对于股票组合，还需计算相关性矩阵，避免过度集中。期货策略应额外关注展期成本、保证金占用和隔夜跳空风险。


# 计算多标的收益率相关性

returns_df = pd.DataFrame()

for ticker in tickers:

    df = pd.read_csv(f'{ticker}_data.csv', index_col=0, parse_dates=True)

    returns_df[ticker] = df['Close'].pct_change()

corr_matrix = returns_df.corr()

print(corr_matrix)

实战中，单一策略往往表现不稳定。可以组合多个策略，如动量策略、均值回归策略、波动率突破策略。利用backtrader或vectorbt等专业回测库，能够更高效地处理订单执行、滑点和手续费。


# 使用vectorbt快速回测多标的（示例）

import vectorbt as vbt

price = vbt.YFData.download(tickers, start='2020-01-01', end='2024-01-01').get('Close')

fast_ma = vbt.MA.run(price, 20)

slow_ma = vbt.MA.run(price, 50)

entries = fast_ma.ma_crossed_above(slow_ma)

exits = fast_ma.ma_crossed_below(slow_ma)

portfolio = vbt.Portfolio.from_signals(price, entries, exits, init_cash=100000, fees=0.001)

print(portfolio.stats())

实盘对接与自动化

批量获取数据并回测后，策略可以对接实盘接口。国内期货可使用CTP接口，股票可使用easytrader或券商API。自动化流程包括定时下载数据、计算信号、发送订单、监控持仓。以下代码展示使用schedule库定时任务。


import schedule

import time

def job():

    # 重新下载数据并计算信号

    data = yf.download(tickers, period='1d')

    for ticker in tickers:

        df = dual_moving_average_strategy(data[ticker])

        last_signal = df['position'].iloc[-1]

        if last_signal == 1:

            print(f'买入信号: {ticker}')

        elif last_signal == -1:

            print(f'卖出信号: {ticker}')

schedule.every().day.at('15:30').do(job)

while True:

    schedule.run_pending()

    time.sleep(60)

注意事项：实盘前需进行样本外测试和模拟交易。期货交易涉及杠杆，风险控制尤为重要。设置止损止盈、仓位限制、最大回撤阈值。定期更新数据，避免使用未来函数。回测中应包含交易成本，股票约0.1%，期货约0.01%。

常见问题与优化方向

数据源稳定性是批量获取的痛点。yfinance可能因网络问题失败，可添加重试机制或备用数据源如akshare、tushare。对于高频数据，需使用pandas的resample方法转换周期。策略优化可尝试参数网格搜索，但需警惕过拟合。


# 参数网格搜索示例

short_windows = [10, 20, 30]

long_windows = [40, 50, 60]

best_sharpe = -np.inf

best_params = None

for sw in short_windows:

    for lw in long_windows:

        if sw >= lw:

            continue

        df = pd.read_csv('AAPL_data.csv', index_col=0, parse_dates=True)

        df = dual_moving_average_strategy(df, sw, lw)

        perf = backtest(df)

        if perf['sharpe'] > best_sharpe:

            best_sharpe = perf['sharpe']

            best_params = (sw, lw)

print(f'最佳参数: {best_params}, 夏普: {best_sharpe}')

量化交易需要持续迭代。结合机器学习方法，如使用scikit-learn预测价格方向，或使用LSTM处理时间序列。风险管理永远优先于收益追求。