第三章：数据源与采集

做量化交易，数据就是你的弹药库。没有好数据，再牛的模型也是白搭。

这一章，我重点聊聊黑色系的数据源怎么选、怎么拿、怎么洗、怎么存。这些都是我这些年踩坑踩出来的经验。

3.1 常用数据源：Mysteel、Wind、SMM

黑色系交易，绕不开这三家数据源。我一个个说。

3.1.1 Mysteel（我的钢铁）

做螺纹钢、热卷、铁矿石的朋友，Mysteel是首选。它的核心数据包括：

• 库存数据：五大品种钢材社会库存、钢厂库存，每周更新
• 产量数据：高炉开工率、电炉开工率、螺纹钢产量
• 表观消费：测算出来的需求数据，市场关注度极高
• 价格指数：全国各地区的现货价格

我个人习惯用Mysteel的周度库存数据做库存周期判断。为什么？因为它连续性好，从2015年到现在几乎没有断过。

3.1.2 Wind（万得）

Wind是金融终端，覆盖面广。黑色系方面，它主要提供：

• 期货行情：螺纹、热卷、铁矿、焦煤、焦炭的日线、分钟线
• 宏观数据：房地产投资、基建投资、PMI等
• 行业数据：粗钢产量、生铁产量（月度）
• 基差数据：期现价差、跨期价差

Wind的API接口叫WindPy，Python可以直接调用。不过要注意，Wind的API有并发限制，我遇到过同时拉太多数据被限流的情况。

3.1.3 SMM（上海有色网）

SMM在有色金属领域是老大，但黑色系方面也有特色数据：

• 铁矿石港口库存：45港库存数据，比Mysteel更细
• 钢厂原料库存：烧结矿、焦炭的可用天数
• 废钢数据：废钢到货量、消耗量

做铁矿石的朋友，我建议把SMM和Mysteel的数据交叉验证。两家口径不同，有时候数据会打架，这时候就要靠经验判断了。

数据源	核心优势	更新频率	API支持
Mysteel	钢材库存、产量	周度	HTTP API
Wind	期货行情、宏观	实时/日/周/月	WindPy
SMM	铁矿石港口库存	周度	HTTP API

3.2 API接口调用

数据拿到了，怎么用代码拉取？我分享几个实战代码片段。

3.2.1 Mysteel API调用示例

Mysteel的API需要申请key，一般走HTTP请求。这是我常用的写法：

import requests
import pandas as pd
from datetime import datetime, timedelta

class MysteelAPI:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.mysteel.com/v1"
    
    def get_inventory_data(self, variety="螺纹钢", date=None):
        """获取库存数据"""
        if date is None:
            date = datetime.now().strftime("%Y-%m-%d")
        
        url = f"{self.base_url}/inventory"
        params = {
            "key": self.api_key,
            "variety": variety,
            "date": date
        }
        
        response = requests.get(url, params=params)
        if response.status_code == 200:
            data = response.json()
            return pd.DataFrame(data["data"])
        else:
            print(f"请求失败: {response.status_code}")
            return None

# 使用示例
api = MysteelAPI("your_api_key_here")
df = api.get_inventory_data("螺纹钢", "2024-01-04")
print(df.head())

3.2.2 Wind API调用示例

WindPy需要先安装，然后启动Wind终端。代码是这样的：

from WindPy import w
import pandas as pd

# 启动Wind
w.start()

# 获取螺纹钢主力合约日线数据
code = "RB.SHF"  # 螺纹钢主力合约
fields = "close,volume,open_interest"
start_date = "2023-01-01"
end_date = "2024-01-01"

# 拉取数据
data = w.wsd(code, fields, start_date, end_date)
df = pd.DataFrame(data.Data, index=data.Fields, columns=data.Times).T

print(df.head())

# 别忘了关闭连接
w.close()

嗯，这里要注意。Wind的wsd函数返回的数据格式有点特殊，需要转成DataFrame。我刚开始用的时候也折腾了半天。

3.3 数据清洗与预处理

数据拉下来，别急着用。原始数据通常有各种问题。我遇到过的情况包括：

• 缺失值：某个日期没有数据
• 异常值：突然跳变，比如库存一天涨了50%
• 重复值：同一日期出现多条记录
• 格式问题：日期格式不统一

这是我的清洗流程：

def clean_inventory_data(df):
    """清洗库存数据"""
    # 1. 处理缺失值
    df = df.dropna(subset=["inventory"])  # 删除库存为空的记录
    df = df.fillna(method="ffill")  # 前向填充其他缺失值
    
    # 2. 处理异常值
    # 用3倍标准差法检测异常
    mean = df["inventory"].mean()
    std = df["inventory"].std()
    df = df[(df["inventory"] > mean - 3*std) & 
            (df["inventory"] < mean + 3*std)]
    
    # 3. 去重
    df = df.drop_duplicates(subset=["date"])
    
    # 4. 统一日期格式
    df["date"] = pd.to_datetime(df["date"])
    df = df.sort_values("date")
    
    return df

# 应用清洗
df_clean = clean_inventory_data(df_raw)

3.4 数据存储方案

数据清洗完了，存哪里？我推荐分层存储：

3.4.1 原始数据层（Raw Data）

存原始API返回的数据，不做任何修改。格式用CSV或Parquet。我习惯按日期分文件夹：

data/
├── raw/
│   ├── mysteel/
│   │   ├── 2023/
│   │   │   ├── 01/
│   │   │   │   ├── inventory_2023-01-05.csv
│   │   │   │   └── price_2023-01-05.csv
│   │   │   └── ...
│   │   └── 2024/
│   └── wind/
│       └── futures/
│           └── RB_2023.csv

3.4.2 清洗数据层（Clean Data）

存清洗后的数据，统一格式。推荐用Parquet格式，压缩率高、读取快：

data/
└── clean/
    ├── inventory.parquet
    ├── price.parquet
    └── macro.parquet

3.4.3 特征数据层（Feature Data）

存计算好的技术指标、库存周期信号等。用HDF5或SQLite：

import pandas as pd

# 计算库存周期信号
def compute_inventory_signal(df):
    """计算库存周期信号"""
    df["inventory_change"] = df["inventory"].pct_change()
    df["price_change"] = df["price"].pct_change()
    
    # 库存周期四象限
    conditions = [
        (df["inventory_change"] > 0) & (df["price_change"] > 0),  # 主动补库
        (df["inventory_change"] > 0) & (df["price_change"] < 0),  # 被动补库
        (df["inventory_change"] < 0) & (df["price_change"] < 0),  # 主动去库
        (df["inventory_change"] < 0) & (df["price_change"] > 0)   # 被动去库
    ]
    choices = [1, 2, 3, 4]
    df["cycle_signal"] = np.select(conditions, choices, default=0)
    
    return df

# 保存特征数据
df_features = compute_inventory_signal(df_clean)
df_features.to_hdf("features.h5", key="inventory_cycle")

3.5 知识体系总览

说了这么多，我画张图帮你理清思路：

这张图把整个数据流程串起来了。从数据源到采集，再到清洗和存储，每一步都有对应的工具和方法。

数据源这块，说白了就是"拿来主义"。但怎么拿、怎么洗、怎么存，决定了你后续策略的成败。记住一句话：垃圾数据进，垃圾策略出。