第1章：数据采集与清洗 — 智能投顾的“原材料”处理

大家好，我是你们的老朋友。今天咱们聊聊智能投顾项目里最基础、也最容易被忽视的一环——数据采集与清洗。

说实话，我在金融科技这行摸爬滚打这么多年，见过太多团队把精力全砸在模型调参上，结果数据一塌糊涂，模型跑出来全是垃圾。嗯，这就是典型的“garbage in, garbage out”。

数据采集与清洗，说白了就是给智能投顾系统准备“食材”。食材不新鲜，再好的厨师也做不出好菜。我个人习惯把这一阶段叫做“数据炼油”——把原油提炼成可用的汽油。

1.1 数据源接入：API、数据库、文件

智能投顾需要的数据，来源五花八门。我把它归纳为三大类：API接口、数据库、文件。

1.1.1 API接口接入

API是现在最主流的数据获取方式。比如获取实时行情、基金净值、宏观经济指标，基本都是通过RESTful API。

我在项目中遇到过一个问题：某家数据供应商的API，每天凌晨2点到3点会做系统维护。如果我们的定时任务正好卡在这个时间段，就会拿不到数据。后来我加了个重试机制，配合指数退避策略，才彻底解决。

import requests
import time

def fetch_market_data(symbol, retries=3):
    url = f"https://api.example.com/v1/quote/{symbol}"
    for attempt in range(retries):
        try:
            resp = requests.get(url, timeout=5)
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            print(f"第{attempt+1}次请求失败: {e}")
            time.sleep(2 ** attempt)  # 指数退避
    return None

1.1.2 数据库接入

内部数据通常存在数据库里。比如客户交易记录、持仓信息、风险评估结果。常用的有MySQL、PostgreSQL、MongoDB。

我个人习惯用ORM框架（比如SQLAlchemy）来操作数据库，而不是直接写裸SQL。为什么？因为ORM能帮你做类型转换、连接池管理，还能防SQL注入。你想想看，金融数据这么敏感，安全第一。

from sqlalchemy import create_engine, text

engine = create_engine("postgresql://user:pass@host:5432/investment_db")
with engine.connect() as conn:
    result = conn.execute(
        text("SELECT client_id, risk_score FROM client_profile WHERE updated_at > :cutoff"),
        {"cutoff": "2024-01-01"}
    )
    for row in result:
        print(row.client_id, row.risk_score)

1.1.3 文件接入

别小看文件接入。很多券商、基金公司还在用Excel或CSV文件交换数据。我接过一个项目，对方每周五下午发一个加密的ZIP包，里面是几十个CSV文件。

处理文件接入，核心是做好格式校验和编码处理。尤其是中文CSV，经常遇到GBK和UTF-8混用的情况。

import pandas as pd

# 读取CSV时自动检测编码
def read_csv_safe(filepath):
    encodings = ['utf-8', 'gbk', 'gb2312', 'latin1']
    for enc in encodings:
        try:
            df = pd.read_csv(filepath, encoding=enc)
            return df
        except UnicodeDecodeError:
            continue
    raise ValueError("无法解码文件，请检查编码格式")

1.2 数据质量检查

数据拿到手，第一件事不是分析，而是检查质量。我总结了一套“五维检查法”：

检查维度	检查内容	常见问题
完整性	字段是否有空值	客户年龄缺失、风险等级为空
准确性	数据是否在合理范围	年龄200岁、收益率500%
一致性	不同来源数据是否匹配	A系统客户ID与B系统不一致
时效性	数据是否最新	用了3个月前的持仓数据
唯一性	是否有重复记录	同一个客户出现多次

为什么要做这么细？我曾经接手过一个项目，客户画像模型跑出来结果很奇怪。查了半天，发现是数据源里有个字段“投资经验”填的全是乱码。嗯，从那以后，我每接一个新数据源，必先跑一遍质量检查脚本。

def data_quality_report(df):
    report = {}
    report['总记录数'] = len(df)
    report['缺失值统计'] = df.isnull().sum().to_dict()
    report['重复记录数'] = df.duplicated().sum()
    # 数值型字段的统计
    num_cols = df.select_dtypes(include=['number']).columns
    for col in num_cols:
        report[f'{col}_最小值'] = df[col].min()
        report[f'{col}_最大值'] = df[col].max()
    return report

1.3 缺失值处理

缺失值，说白了就是数据里有些格子是空的。处理方式取决于业务场景。

我常用的三种方法：

• 删除法：如果缺失比例很小（比如<5%），直接删掉那几行。简单粗暴，但要注意别删太多导致样本不足。
• 填充法：用均值、中位数、众数填充。比如客户年龄缺失，可以用整体客户的平均年龄填充。
• 模型预测法：用其他字段预测缺失值。比如用收入、职业预测风险等级。这个方法最准，但成本也最高。

# 按客户分层填充缺失值
def fill_missing_by_group(df, group_col, target_col):
    grouped_means = df.groupby(group_col)[target_col].transform('median')
    df[target_col].fillna(grouped_means, inplace=True)
    return df

1.4 异常值检测与处理

异常值，就是那些明显不合理的数据。比如客户年龄200岁，或者某只股票一天涨了1000%。

检测方法我常用两种：

• 3σ原则：数据服从正态分布时，超过均值±3倍标准差的值视为异常。
• 箱线图法：用四分位数判断，超出上下须的数据就是异常。这个方法不受极端值影响，更稳健。

处理方式呢？我一般分三步走：

1. 先标记异常值，不要直接删除
2. 人工核查，看是不是数据录入错误
3. 如果是错误，修正或删除；如果是真实值（比如黑天鹅事件），保留但做特殊标记

def detect_outliers_iqr(df, column):
    Q1 = df[column].quantile(0.25)
    Q3 = df[column].quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    outliers = df[(df[column] < lower_bound) | (df[column] > upper_bound)]
    return outliers

本章知识体系总览

下面这张图，是我自己画的本章知识结构。你可以把它当成一张“地图”，随时回来对照。

好了，这一章的内容就到这里。数据采集与清洗，听起来枯燥，但它是整个智能投顾系统的地基。地基打不牢，上面盖的楼再漂亮也得塌。

下一章我们会深入聊聊客户画像的特征工程。不过在那之前，我建议你把今天讲的代码跑一遍，亲手处理一份真实的数据。相信我，踩过坑才能记得牢。

---