避开Yahoo Finance反爬！用Alpha Vantage API轻松获取苹果股票10年数据（附完整Python代码）

从数据获取到模型部署：构建一个面向未来的美股量化分析系统

最近在尝试复现一些经典的Kaggle时间序列预测项目时，很多朋友都遇到了一个共同的拦路虎：数据源。那些几年前还能顺畅运行的yfinance脚本，如今频繁遭遇请求限制，数据获取变得异常困难。这不仅仅是技术上的小麻烦，它直接关系到整个分析项目的起点是否稳固。对于量化交易初学者和数据分析师而言，一个稳定、可靠的数据管道，其重要性不亚于模型算法本身。今天，我们不谈那些已经被“反爬”机制层层设防的免费公开接口，而是转向一个更专业、更稳定的替代方案，并在此基础上，构建一套从数据获取、处理到LSTM模型预测的完整工作流。我们将以苹果公司（AAPL）的股票数据为例，手把手带你搭建一个属于自己的、可扩展的金融数据分析系统。

1. 构建稳健的金融数据管道：告别不稳定源

在数据科学和量化分析领域，有一句老话：垃圾进，垃圾出。如果你的数据源头本身就不稳定、不完整，那么后续无论构建多么精妙的模型，其结论都可能是空中楼阁。过去，许多教程和项目依赖雅虎财经等免费接口，但随着这些平台对自动化请求的限制日益严格，频繁的rate limiting和连接中断成了家常便饭。

1.1 为什么选择Alpha Vantage作为核心数据源？

Alpha Vantage提供了一个相对友好且功能强大的免费层API。与一些完全开放的接口相比，它通过API密钥进行管理，这意味着请求更可控，服务也相对稳定。其核心优势在于：

• 结构化数据：返回的数据是标准的JSON或Pandas DataFrame格式，包含开盘价、最高价、最低价、收盘价、成交量等关键字段，无需复杂的HTML解析。
• 历史数据完整：免费层通常提供长达20年的日线历史数据，对于大多数回溯测试和模型训练来说已经足够。
• 明确的频率限制：虽然免费API有每分钟/每日的调用次数限制，但规则清晰，便于我们在代码中设计合理的延时策略来规避。
• 丰富的资产类别：不仅覆盖美股，还包括外汇、加密货币和大宗商品等，为策略的多元化提供了可能。

当然，它也有其局限性。免费API不提供调整后收盘价，这对于长期趋势分析来说是一个需要考虑的因素。公司行动如分红、拆股会影响股价的连续性，调整后价格则消除了这些非市场因素的影响，使得不同时期的价格更具可比性。对于学习和小型项目，我们可以暂时使用收盘价；但对于严肃的策略回测，寻找或计算调整后价格是必要的一步。

1.2 实战：自动化获取与保存多只股票数据

下面，我们构建一个Python脚本，用于批量获取并保存多只科技股的历史数据。这个脚本的核心是处理API限速，并确保数据的持久化存储。

首先，你需要前往Alpha Vantage官网注册一个免费账户，获取你的API密钥。

# data_pipeline.py import os import time from datetime import datetime, timedelta import pandas as pd from alpha_vantage.timeseries import TimeSeries # 配置部分 API_KEY = 'YOUR_ALPHA_VANTAGE_API_KEY' # 请替换为你的实际密钥 SYMBOL_LIST = ['AAPL', 'MSFT', 'GOOGL', 'AMZN'] # 目标股票代码列表 DATA_DIR = './stock_data' # 数据保存目录 YEARS_OF_HISTORY = 10 # 获取多少年的历史数据 # 创建数据目录 os.makedirs(DATA_DIR, exist_ok=True) # 初始化TimeSeries客户端 ts = TimeSeries(key=API_KEY, output_format='pandas', indexing_type='date') print(f"开始获取近{YEARS_OF_HISTORY}年股票数据...") start_date_cutoff = datetime.now() - timedelta(days=365 * YEARS_OF_HISTORY) for symbol in SYMBOL_LIST: file_path = os.path.join(DATA_DIR, f"{symbol}_daily.csv") # 检查文件是否已存在，避免重复下载 if os.path.exists(file_path): print(f"文件已存在，跳过 {symbol}") continue print(f"正在获取 {symbol} 的数据...") try: # 获取全量日线数据 data, meta_data = ts.get_daily(symbol=symbol, outputsize='full') # 确保索引是日期时间类型并按日期排序 data.index = pd.to_datetime(data.index) data = data.sort_index() # 筛选指定年份之后的数据 data = data[data.index >= start_date_cutoff] # 重命名列，使其更易读 data.rename(columns={ '1. open': 'Open', '2. high': 'High', '3. low': 'Low', '4. close': 'Close', '5. volume': 'Volume' }, inplace=True) # 保存到CSV data.to_csv(file_path) print(f"成功保存 {symbol} 数据至 {file_path}，共 {len(data)} 条记录。") except Exception as e: print(f"获取 {symbol} 时出错: {e}") # 严格遵守API调用频率限制，免费版建议间隔至少12秒 time.sleep(15) print("所有股票数据获取完成！")

注意：将YOUR_ALPHA_VANTAGE_API_KEY替换为你自己的密钥。time.sleep(15)是遵守免费API每分钟5次调用限制的安全做法。如果你的需求更大，可以考虑购买其付费套餐。

运行这个脚本后，你的./stock_data目录下会生成类似AAPL_daily.csv的文件。用Pandas打开看一眼，数据结构清晰，包含日期索引和OHLCV数据，为下一步分析打下了坚实基础。

2. 数据预处理：为时间序列模型准备“食材”

拿到原始数据只是第一步。金融时间序列数据通常存在噪声、非平稳性等问题，直接喂给模型效果往往不佳。预处理的目标是将其转化为模型易于学习和理解的格式。

2.1 关键预处理步骤解析

我们以苹果公司（AAPL）的数据为例，演示完整的预处理流程。

# data_preprocessing.py import pandas as pd import numpy as np from sklearn.preprocessing import MinMaxScaler def load_and_preprocess(symbol, data_dir='./stock_data'): """加载并预处理单只股票数据""" file_path = f'{data_dir}/{symbol}_daily.csv' df = pd.read_csv(file_path, index_col=0, parse_dates=True) # 1. 确保数据按日期排序 df = df.sort_index() # 2. 选择我们关心的特征：收盘价 # 注意：这里使用的是未调整的收盘价('Close')。对于长期分析，应考虑寻找调整后收盘价数据源。 df = df[['Close']].copy() # 3. 处理缺失值（金融数据周末休市，但我们的CSV是日线，通常无缺失） if df['Close'].isnull().any(): # 前向填充或插值 df['Close'].fillna(method='ffill', inplace=True) # 4. 可视化原始数据（可选，但建议） import matplotlib.pyplot as plt plt.figure(figsize=(12, 6)) plt.plot(df.index, df['Close'], label=f'{symbol} Close Price') plt.title(f'{symbol} Historical Closing Price') plt.xlabel('Date') plt.ylabel('Price (USD)') plt.legend() plt.grid(True) plt.tight_layout() plt.savefig(f'{symbol}_raw_price.png', dpi=150) plt.close() return df # 执行预处理 symbol = 'AAPL' df_aapl = load_and_preprocess(symbol) print(f"数据时间范围: {df_aapl.index.min()} 到 {df_aapl.index.max()}") print(f"数据总条数: {len(df_aapl)}")

接下来是最关键的一步：为LSTM模型创建监督学习数据集。LSTM是循环神经网络的一种，擅长处理序列数据。我们需要将时间序列数据构造成(样本数, 时间步长, 特征数)的格式。

def create_sequences(data, sequence_length=60): """ 将时间序列数据转换为LSTM所需的序列样本。 参数: data: 一维数组，例如收盘价序列。 sequence_length: 用过去多少天的数据来预测下一天。 返回: X: 特征序列，形状为 (n_samples, sequence_length, 1) y: 目标值，形状为 (n_samples,) """ X, y = [], [] for i in range(sequence_length, len(data)): X.append(data[i-sequence_length:i]) # 过去sequence_length天的数据 y.append(data[i]) # 第i天的数据（我们要预测的） return np.array(X), np.array(y) # 数据归一化：将价格缩放到0-1之间，加速模型收敛 scaler = MinMaxScaler(feature_range=(0, 1)) scaled_prices = scaler.fit_transform(df_aapl[['Close']]) # 创建序列 SEQ_LEN = 60 # 使用过去60天的数据 X, y = create_sequences(scaled_prices, SEQ_LEN) # 划分训练集和测试集 (例如，95%训练，5%测试) split_idx = int(len(X) * 0.95) X_train, X_test = X[:split_idx], X[split_idx:] y_train, y_test = y[:split_idx], y[split_idx:] print(f"训练集样本数: {X_train.shape[0]}") print(f"测试集样本数: {X_test.shape[0]}") print(f"每个样本的形状 (时间步长, 特征): {X_train.shape[1:]}")

至此，数据已经准备好了。我们有了形状为(n_samples, 60, 1)的训练特征X_train，和对应的目标值y_train。这个“60天预测下一天”的设定，是时间序列预测中一个常见的滑动窗口方法。

3. 构建与训练LSTM预测模型

有了高质量的数据，我们就可以构建模型了。这里我们使用PyTorch框架来实现一个多层LSTM网络。PyTorch的动态图机制使得模型构建和调试非常灵活。

3.1 定义LSTM网络结构

我们的模型将是一个相对标准的回归模型：输入过去60天的价格序列，输出对第61天价格的预测。

# model_definition.py import torch import torch.nn as nn class StockPriceLSTM(nn.Module): def __init__(self, input_size=1, hidden_size=64, num_layers=2, output_size=1): super(StockPriceLSTM, self).__init__() self.hidden_size = hidden_size self.num_layers = num_layers # LSTM层 self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True, dropout=0.2) # Dropout层防止过拟合 self.dropout = nn.Dropout(0.3) # 全连接输出层 self.linear = nn.Linear(hidden_size, output_size) def forward(self, x): # 初始化隐藏状态和细胞状态 h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device) c0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device) # LSTM前向传播 out, _ = self.lstm(x, (h0, c0)) # 我们只取最后一个时间步的输出 out = out[:, -1, :] # 应用Dropout out = self.dropout(out) # 全连接层得到最终预测 out = self.linear(out) return out

这个模型结构有几个可调的超参数：

• hidden_size: LSTM单元隐藏状态的维度，决定了模型的容量。
• num_layers: 堆叠的LSTM层数，增加深度可能捕捉更复杂的模式，但也更容易过拟合。
• dropout: 随机丢弃部分神经元，是防止模型在训练集上过拟合的有效正则化手段。

3.2 模型训练与评估

接下来，我们将数据转换为PyTorch张量，并开始训练循环。

# model_training.py import torch.optim as optim from torch.utils.data import DataLoader, TensorDataset # 检查GPU是否可用 device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') print(f"使用设备: {device}") # 将数据转换为PyTorch张量并移至设备 X_train_tensor = torch.FloatTensor(X_train).to(device) y_train_tensor = torch.FloatTensor(y_train).to(device) X_test_tensor = torch.FloatTensor(X_test).to(device) y_test_tensor = torch.FloatTensor(y_test).to(device) # 创建DataLoader，方便批量训练 train_dataset = TensorDataset(X_train_tensor, y_train_tensor) train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True) # 初始化模型、损失函数和优化器 model = StockPriceLSTM().to(device) criterion = nn.MSELoss() # 均方误差损失，适用于回归问题 optimizer = optim.Adam(model.parameters(), lr=0.001) # 训练循环 num_epochs = 200 train_losses = [] test_losses = [] for epoch in range(num_epochs): model.train() epoch_train_loss = 0 for batch_X, batch_y in train_loader: optimizer.zero_grad() outputs = model(batch_X) loss = criterion(outputs, batch_y) loss.backward() # 梯度裁剪，防止梯度爆炸（在RNN中常见） torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() epoch_train_loss += loss.item() avg_train_loss = epoch_train_loss / len(train_loader) train_losses.append(avg_train_loss) # 在测试集上评估 model.eval() with torch.no_grad(): test_outputs = model(X_test_tensor) test_loss = criterion(test_outputs, y_test_tensor) test_losses.append(test_loss.item()) if (epoch + 1) % 20 == 0: print(f'Epoch [{epoch+1}/{num_epochs}], Train Loss: {avg_train_loss:.6f}, Test Loss: {test_loss.item():.6f}') print("训练完成！")

在训练过程中，观察训练损失和测试损失的变化至关重要。理想情况下，两者应同步下降并逐渐趋于平稳。如果训练损失持续下降而测试损失开始上升，则是过拟合的典型信号，此时需要增加Dropout比率、获取更多数据或使用更早停止的策略。

4. 预测、可视化与系统化思考

模型训练完成后，我们不仅要看损失函数，更要看它在实际价格曲线上的预测表现。

4.1 生成预测并反归一化

# prediction_and_visualization.py import matplotlib.pyplot as plt # 使用模型进行预测 model.eval() with torch.no_grad(): train_predict = model(X_train_tensor).cpu().numpy() test_predict = model(X_test_tensor).cpu().numpy() # 将归一化的预测值转换回原始价格区间 train_predict = scaler.inverse_transform(train_predict) y_train_actual = scaler.inverse_transform(y_train.reshape(-1, 1)) test_predict = scaler.inverse_transform(test_predict) y_test_actual = scaler.inverse_transform(y_test.reshape(-1, 1)) # 为预测值创建对应的时间索引 # 注意：由于我们使用滑动窗口，预测值的时间点比原始数据晚了SEQ_LEN天 train_predict_index = df_aapl.index[SEQ_LEN:SEQ_LEN + len(train_predict)] test_predict_index = df_aapl.index[SEQ_LEN + len(train_predict): SEQ_LEN + len(train_predict) + len(test_predict)] # 绘制结果 plt.figure(figsize=(16, 8)) plt.plot(df_aapl.index, df_aapl['Close'], label='Actual Price', alpha=0.7, color='gray') plt.plot(train_predict_index, train_predict, label='Train Prediction', linewidth=2) plt.plot(test_predict_index, test_predict, label='Test Prediction', linewidth=2, color='red') plt.title(f'{symbol} Stock Price Prediction using LSTM') plt.xlabel('Date') plt.ylabel('Close Price (USD)') plt.legend() plt.grid(True, alpha=0.3) plt.tight_layout() plt.savefig(f'{symbol}_lstm_prediction.png', dpi=150) plt.show()

这张图能直观地告诉我们模型学到了什么。如果测试集（红色部分）的预测曲线能大致跟随真实价格的波动趋势，说明模型具备一定的泛化能力。但请务必清醒：股价预测是金融领域最难的挑战之一，受无数不可预测因素影响。我们的LSTM模型捕捉的更多是历史数据中的局部模式和趋势，绝不能将其视为稳赚不赔的交易信号。

4.2 超越单步预测：尝试序列到序列（Seq2Seq）模型

上面的模型是“单步预测”，即用过去N天预测下一天。但有时我们想直接预测未来多天（例如未来30天）的走势。这就需要用到Seq2Seq结构的LSTM。

其核心思想是使用一个编码器（Encoder）LSTM将输入序列压缩为一个上下文向量，再用一个解码器（Decoder）LSTM根据该向量一步步生成输出序列。在股价预测中，解码器每一步的输入可以是上一步的预测值（自回归），逐步推演出未来序列。

# 简化的Seq2Seq模型思路（代码框架） class Seq2SeqLSTM(nn.Module): def __init__(self, input_size=1, hidden_size=128, num_layers=2, output_steps=30): super().__init__() self.encoder = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True) self.decoder = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True) self.fc = nn.Linear(hidden_size, 1) self.output_steps = output_steps def forward(self, src): # src: [batch, seq_len, features] batch_size = src.shape[0] # 编码 _, (hidden, cell) = self.encoder(src) # 解码器初始输入（例如，源序列的最后一个值） decoder_input = src[:, -1:, :] outputs = [] for _ in range(self.output_steps): out, (hidden, cell) = self.decoder(decoder_input, (hidden, cell)) pred = self.fc(out[:, -1, :]) outputs.append(pred.unsqueeze(1)) # 将本次预测作为下一次解码的输入（自回归） decoder_input = pred.unsqueeze(1) return torch.cat(outputs, dim=1)

实现并训练Seq2Seq模型比单步预测更复杂，需要精心设计损失函数（如对输出序列的每个点计算损失）和训练策略。但它能提供一个对未来一段时间走势的整体展望，在场景分析中可能更有用。

4.3 构建一个可维护的系统

最后，让我们跳出单次脚本的思维，思考如何将这一切系统化。一个健壮的量化分析系统应该包含以下模块：

例如，一个简单的模型服务化脚本可能长这样：

# app.py (FastAPI示例) from fastapi import FastAPI import torch import numpy as np from pydantic import BaseModel from model_definition import StockPriceLSTM # 导入之前定义的模型 from data_preprocessing import scaler # 导入之前拟合好的归一化器 app = FastAPI() model = StockPriceLSTM() model.load_state_dict(torch.load('best_model.pth')) model.eval() class PredictionRequest(BaseModel): past_60_days_prices: list[float] # 过去60天的收盘价列表 @app.post("/predict/") def predict_next_price(request: PredictionRequest): # 1. 预处理输入 prices_array = np.array(request.past_60_days_prices).reshape(-1, 1) scaled_prices = scaler.transform(prices_array) input_tensor = torch.FloatTensor(scaled_prices).unsqueeze(0) # 增加batch维度 # 2. 预测 with torch.no_grad(): prediction_scaled = model(input_tensor).numpy() # 3. 反归一化 next_price = scaler.inverse_transform(prediction_scaled.reshape(-1, 1))[0, 0] return {"predicted_next_close_price": round(float(next_price), 2)}

这样一来，你的前端或其他系统就可以通过HTTP请求轻松获取模型的预测结果，整个分析流程就从一次性的脚本，升级成了一个可重复使用、可扩展的微服务。

走完这一整套流程，你收获的不仅仅是一个能预测股价的LSTM模型，更是一套应对金融数据获取挑战的工程化解决方案和系统化思维。从被动的“爬虫-反爬”对抗，转向主动利用稳定的商业API；从零散的Jupyter Notebook，转向模块化、可维护的代码库。这才是应对快速变化的技术环境时，真正有价值的能力提升。在实际项目中，我通常会为数据管道设置独立的日志和监控，确保任何一环出错都能第一时间知晓，而不是等到模型训练失败时才回头检查。金融数据瞬息万变，系统的可靠性永远是第一位的。