第三章：数据仓库与特征工程：数据存储方案、特征提取、选择与降维、质量监控

大家好，我是老张。今天咱们聊聊投研流程里最“脏”最“累”的活——数据仓库与特征工程。

说实话，很多做AI投研的朋友，一上来就喜欢调模型、调参数。但我个人经验是，模型再牛，也救不了垃圾数据。你想想看，如果喂进去的数据都是错的，或者特征之间高度冗余，那模型学出来的东西，说白了就是“精致的胡说八道”。

所以，这一章我带你走一遍数据从存储到特征落地的完整链路。我会结合我踩过的坑，把SQL/NoSQL怎么选、特征怎么提、怎么降维、怎么监控质量，一次性讲透。

3.1 数据存储方案：SQL vs NoSQL，别选错了

先聊存储。投研数据有个特点：既有高度结构化的行情数据，又有非结构化的新闻、研报、舆情文本。

我个人习惯是：结构化数据用SQL，非结构化或半结构化数据用NoSQL。但这不是绝对的，得看场景。

3.1.1 SQL 方案：PostgreSQL + TimescaleDB

我在项目中遇到过，早期用MySQL存分钟级K线数据，结果表一大了，查询就慢得像蜗牛。后来换成了PostgreSQL + TimescaleDB，才算是解决了。

TimescaleDB是PostgreSQL的一个扩展，专门处理时序数据。它自动按时间分区，查询性能提升明显。

-- 创建超表（Hypertable）
CREATE TABLE stock_ohlc (
    time        TIMESTAMPTZ NOT NULL,
    symbol      TEXT NOT NULL,
    open        DOUBLE PRECISION,
    high        DOUBLE PRECISION,
    low         DOUBLE PRECISION,
    close       DOUBLE PRECISION,
    volume      BIGINT
);

-- 转换为超表，按时间和股票代码分区
SELECT create_hypertable('stock_ohlc', 'time', chunk_time_interval => INTERVAL '1 day');

-- 查询某只股票最近一周的收盘价
SELECT time, close
FROM stock_ohlc
WHERE symbol = '000001.SZ'
  AND time >= NOW() - INTERVAL '7 days'
ORDER BY time;

嗯，这里要注意：chunk_time_interval 要根据你的数据量来调。我一般设成1天到7天，太短了分区太多，太长了查询性能下降。

3.1.2 NoSQL 方案：MongoDB 与 Elasticsearch

对于新闻、研报这种文本数据，我推荐MongoDB。它的文档模型很灵活，字段可以随时加，不用改表结构。

举个例子，一条新闻可能有标题、正文、发布时间、情感标签、关联股票列表。用MongoDB存，就是一个JSON文档，非常自然。

// MongoDB 文档示例
{
  "_id": ObjectId("..."),
  "title": "某公司发布2024年业绩预告",
  "content": "公司预计净利润同比增长50%...",
  "publish_time": ISODate("2024-01-15T09:30:00Z"),
  "sentiment": 0.85,
  "related_stocks": ["000001.SZ", "600000.SH"]
}

另外，Elasticsearch 我主要用来做全文搜索。投研人员经常要搜“某公司 业绩 暴雷”这种关键词，ES的倒排索引比SQL的LIKE快几个数量级。

3.2 特征提取方法：从原始数据到有效特征

数据存好了，下一步就是提取特征。特征提取，说白了就是把原始数据变成模型能理解的数值。

我把它分成三类：时序特征、文本特征、另类数据特征。

3.2.1 时序特征提取

股票价格、成交量这些时序数据，最常用的特征包括：

• 统计特征：过去N天的均值、标准差、最大值、最小值、分位数
• 技术指标：RSI、MACD、布林带、移动平均线
• 波动率特征：已实现波动率、隐含波动率
• 时序交叉特征：不同时间窗口的均值之差（如5日均线-20日均线）

我记得有一次做因子回测，发现一个简单的“过去5天涨幅”特征，居然比一堆复杂的技术指标效果还好。有时候，简单就是美。

import pandas as pd
import numpy as np

def extract_time_series_features(df, window=5):
    """
    提取时序统计特征
    df: 包含 'close' 列的 DataFrame
    """
    df = df.copy()
    # 过去5天的均值
    df['ma_5'] = df['close'].rolling(window=window).mean()
    # 过去5天的标准差
    df['std_5'] = df['close'].rolling(window=window).std()
    # 过去5天的涨跌幅
    df['return_5'] = df['close'].pct_change(periods=window)
    # 过去5天的最大值与最小值之差
    df['range_5'] = df['close'].rolling(window=window).max() - df['close'].rolling(window=window).min()
    return df

3.2.2 文本特征提取

新闻、研报这些文本数据，怎么变成特征？我常用的方法：

• TF-IDF：简单有效，适合关键词提取
• Word2Vec / GloVe：词向量，能捕捉语义相似性
• BERT 等预训练模型：效果最好，但计算成本高
• 情感分析：用预训练模型或词典，输出情感得分

3.3 特征选择与降维：别让特征“打架”

特征多了，不一定好。你想想看，如果两个特征高度相关（比如“开盘价”和“收盘价”），模型就会学偏，还容易过拟合。

特征选择，就是去掉冗余的、无关的特征。降维，则是把高维特征压缩到低维空间。

3.3.1 特征选择方法

方法	原理	适用场景
方差过滤	去掉方差接近0的特征	特征值几乎不变的情况
相关系数法	计算特征与目标变量的相关性	线性关系明显的场景
卡方检验	检验特征与目标变量的独立性	分类问题，特征为离散值
基于模型的选择	用随机森林/Lasso等模型筛选重要特征	非线性关系，特征数量较多

我个人最常用的是基于模型的选择。用Lasso回归，它会自动把不重要的特征系数压缩到0，非常直观。

from sklearn.linear_model import Lasso
from sklearn.feature_selection import SelectFromModel

# 假设 X 是特征矩阵，y 是目标变量
lasso = Lasso(alpha=0.01)
lasso.fit(X, y)

# 选择系数不为0的特征
selector = SelectFromModel(lasso, prefit=True)
X_selected = selector.transform(X)

# 查看哪些特征被选中
selected_features = X.columns[selector.get_support()]
print(f"选中的特征: {selected_features.tolist()}")

3.3.2 降维：PCA 与 t-SNE

降维最经典的是PCA（主成分分析）。它把原始特征投影到几个“主成分”上，保留最大方差。

但PCA有个缺点：解释性差。你很难说清楚“主成分1”到底代表什么。

所以，如果是为了可视化，我更喜欢用t-SNE。它能很好地保持数据的局部结构，画出来的图很漂亮。

3.4 数据质量监控：别让坏数据毁了模型

数据质量，是投研的命门。我见过太多案例，因为数据质量没监控，模型上线后突然崩了。

数据质量监控，我总结为“四看”：

• 看完整性：有没有缺失值？缺失率是否超过阈值？
• 看准确性：数值是否在合理范围内？比如股价不可能为负数。
• 看一致性：不同数据源之间的数据是否矛盾？比如A股代码在Wind和Tushare里是否一致？
• 看时效性：数据是否及时更新？延迟超过多久需要告警？

3.4.1 自动化监控脚本

我习惯写一个简单的监控脚本，每天跑一次。一旦发现异常，就发邮件或钉钉告警。

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

def check_data_quality(df, symbol_col='symbol', time_col='time', price_col='close'):
    """
    数据质量检查
    """
    issues = []
    
    # 1. 检查缺失值
    missing_rate = df.isnull().mean()
    for col, rate in missing_rate.items():
        if rate > 0.05:  # 缺失率超过5%就告警
            issues.append(f"列 {col} 缺失率 {rate:.2%}")
    
    # 2. 检查价格是否合理
    if price_col in df.columns:
        invalid_prices = df[df[price_col] <= 0]
        if len(invalid_prices) > 0:
            issues.append(f"发现 {len(invalid_prices)} 条价格为0或负数的记录")
    
    # 3. 检查数据时效性
    latest_time = df[time_col].max()
    if latest_time < datetime.now() - timedelta(hours=1):
        issues.append(f"数据最新时间 {latest_time}，已超过1小时未更新")
    
    return issues

# 使用示例
issues = check_data_quality(df)
if issues:
    for issue in issues:
        print(f"[告警] {issue}")
    # 发送告警邮件或钉钉消息
else:
    print("数据质量检查通过")

3.5 本章知识体系图

下面这张图，帮你梳理本章的核心逻辑。从数据存储，到特征提取，再到特征选择与降维，最后是质量监控，环环相扣。

好了，这一章的内容就到这里。数据存储和特征工程，是投研的“地基”。地基打不牢，楼盖得再高也得塌。希望你能把这些方法用起来，少踩一些我当年踩过的坑。

---

公众号:蓝海数据掘金营，微信deep3321