企业微信

批量获取股票数据

在量化交易中，数据是策略构建的基础。Python 提供了多种工具用于从公开金融接口抓取股票历史行情。yfinance 是一个广泛使用的库，能够免费访问 Yahoo Finance 的数据源，支持多只股票的同时下载。

安装依赖库：


pip install yfinance pandas numpy matplotlib

使用以下代码一次性获取多个股票的历史价格数据：

如何用Python批量获取股票数据并实现交易策略


import yfinance as yf

import pandas as pd

# 定义目标股票列表

tickers = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'TSLA']

# 设置时间范围

start_date = '2020-01-01'

end_date = '2024-12-31'

# 批量下载数据

data = yf.download(tickers, start=start_date, end=end_date)['Adj Close']

# 保存为CSV文件便于后续处理

data.to_csv('stock_prices.csv')

print("股票数据已成功下载并保存")

该方法返回一个包含各股票复权收盘价的 DataFrame，列名为股票代码，行为日期索引。这种结构适合进行向量化运算和跨资产比较。

对于需要更高频或更完整字段（如财务指标、期权链）的场景，可接入 akshare 或 tushare 等国内金融数据接口。以 A 股为例：


import akshare as ak

# 获取沪深A股代码列表

stock_list = ak.stock_zh_a_spot_em()

# 提取前5只股票进行数据采集

target_stocks = stock_list['代码'][:5].tolist()

price_data = {}

for symbol in target_stocks:

    try:

        df = ak.stock_zh_a_hist(symbol=symbol, period="daily", adjust="hfq")

        price_data[symbol] = df.set_index('日期')['收盘']

    except Exception as e:

        print(f"获取 {symbol} 数据失败: {e}")

此方式适用于中国市场的本地化数据需求，尤其在处理除权除息调整方面表现稳定。

构建基础交易信号

获取原始数据后，下一步是生成可执行的买卖信号。常用技术指标包括移动平均线、布林带和RSI。以双均线交叉策略为例，计算每只股票的短期（20日）与长期（60日）均线关系。


import numpy as np

def generate_ma_signal(prices, short_window=20, long_window=60):

    ma_short = prices.rolling(short_window).mean()

    ma_long = prices.rolling(long_window).mean()



    # 生成买入信号：短期均线上穿长期均线

    signals = np.where((ma_short > ma_long) & (ma_short.shift(1) <= ma_long.shift(1)), 1, 0)

    return pd.Series(signals, index=prices.index)

# 对所有股票应用策略

signals_dict = {}

for ticker in data.columns:

    signals_dict[ticker] = generate_ma_signal(data[ticker])

signals_df = pd.DataFrame(signals_dict)

上述函数输出每日是否持有该股票的二元信号（1 表示买入，0 表示卖出），可用于回测仓位变化。

引入波动率过滤机制

单纯依赖趋势可能产生过多假信号，尤其是在震荡市中。引入波动率控制可有效降低交易频率并规避高风险时段。

计算历史波动率通常采用对数收益率的标准差：


def calculate_volatility(returns, window=20):

    log_returns = np.log(returns / returns.shift(1))

    volatility = log_returns.rolling(window).std() * np.sqrt(252)  # 年化波动率

    return volatility

# 添加波动率筛选条件

volatility_dict = {}

filtered_signals = {}

for ticker in data.columns:

    price_series = data[ticker].dropna()

    vol = calculate_volatility(price_series)

    raw_signal = signals_dict[ticker]



    # 当波动率低于阈值时取消交易信号

    final_signal = np.where(vol < 0.3, raw_signal, 0)  # 设定年化波动率阈值为30%

    filtered_signals[ticker] = final_signal

filtered_signals_df = pd.DataFrame(filtered_signals)

该逻辑确保仅在市场波动适中时执行交易，避免在极端行情下频繁进出。

回测策略绩效

完成信号生成后，需评估其实际盈利能力。构造简单回测框架，模拟资金分配与收益累计过程。


def backtest_strategy(prices_df, signals_df, initial_capital=1_000_000):

    portfolio = pd.DataFrame(index=prices_df.index)

    portfolio['total'] = initial_capital

    portfolio['holdings'] = 0

    portfolio['cash'] = initial_capital

    for i in range(1, len(portfolio)):

        daily_signals = signals_df.iloc[i]

        current_prices = prices_df.iloc[i]

        previous_prices = prices_df.iloc[i - 1]

        shares_to_buy = 0

        for ticker in prices_df.columns:

            if daily_signals[ticker] == 1 and portfolio['cash'][i - 1] > current_prices[ticker]:

                num_shares = int(portfolio['cash'][i - 1] / len(prices_df.columns) / current_prices[ticker])

                shares_to_buy += num_shares * current_prices[ticker]

        portfolio['holdings'][i] = shares_to_buy

        portfolio['cash'][i] = portfolio['cash'][i - 1] - shares_to_buy

        portfolio['total'][i] = portfolio['holdings'][i] + portfolio['cash'][i]

    return portfolio

result_portfolio = backtest_strategy(data, filtered_signals_df)

绘制净值曲线直观展示策略表现：


import matplotlib.pyplot as plt

plt.figure(figsize=(12, 6))

plt.plot(result_portfolio['total'], label='Portfolio Value', color='blue')

plt.title('Strategy Equity Curve')

plt.xlabel('Date')

plt.ylabel('Value (USD)')

plt.legend()

plt.grid(True)

plt.show()

结果可进一步计算年化收益率、最大回撤、夏普比率等关键指标，辅助决策优化。

实现自动化运行流程

将整个流程封装为可调度任务，支持定时执行与实时监控。利用 schedule 模块设置每日收盘后自动拉取最新数据并更新信号。


import schedule

import time

def daily_update():

    print("开始更新股票数据...")

    new_data = yf.download(tickers, period="1d", interval="1d")

    latest_close = new_data['Adj Close'].iloc[-1]

    print(f"最新收盘价: {latest_close.to_dict()}")

schedule.every().monday.at("16:00").do(daily_update)

schedule.every().tuesday.at("16:00").do(daily_update)

schedule.every().wednesday.at("16:00").do(daily_update)

schedule.every().thursday.at("16:00").do(daily_update)

schedule.every().friday.at("16:00").do(daily_update)

while True:

    schedule.run_pending()

    time.sleep(60)

配合邮件通知或微信推送模块，可在触发新信号时及时提醒操作。

扩展至期货市场

相同架构亦适用于期货品种。通过 ccxt 库连接交易所API获取合约行情，或使用 vn.py 框架对接国内期货平台。

例如获取比特币永续合约数据：


import ccxt

exchange = ccxt.binance()

ohlcv = exchange.fetch_ohlcv('BTC/USDT', timeframe='1d', limit=1000)

btc_df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])

btc_df['timestamp'] = pd.to_datetime(btc_df['timestamp'], unit='ms')

随后应用相同的均线策略与风险管理规则，形成跨市场统一交易逻辑。

Python 在金融数据分析中的灵活性使其成为构建全自动量化系统的理想选择。从数据采集到策略部署，每个环节均可程序化控制，显著提升投资效率与纪律性。