📚 课程目录第2章逐笔成交流数据清洗与特征提取

2. 逐笔成交流数据清洗与特征提取

任务目标

本案例处理一份模拟的逐笔成交记录（Tick-by-Tick Trade Data），完成两类工作：数据清洗（去重、剔除异常价格与零成交量记录、修正缺失字段）与特征提取（聚合生成每秒的交易量、价格变动、主动买卖方向）。输入为CSV格式的原始逐笔数据，输出为清洗后的明细表与按秒聚合的特征表。

数据说明与依赖

使用Python标准库csv与datetime完成全部逻辑，无需第三方依赖。输入文件trades_raw.csv包含以下字段：

timestamp：格式YYYY-MM-DD HH:MM:SS.mmm（毫秒精度）
price：成交价格（浮点数）
volume：成交量（整数，股/张）
side：主动方向，B（主动买）或S（主动卖），缺失时为空字符串

清洗规则：删除完全重复的行；剔除price <= 0或volume <= 0的记录；若side缺失但价格高于上一笔有效成交价，则标记为B，低于则标记为S，等于则沿用上一笔方向（首笔缺失且无法判断时丢弃）。

完整实现

import csv
from datetime import datetime

def clean_and_aggregate(input_path, output_detail, output_feature):
    seen = set()
    cleaned = []
    last_price = None
    last_side = None

    with open(input_path, 'r', newline='', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            ts = row['timestamp'].strip()
            price = float(row['price'])
            volume = int(row['volume'])
            side = row['side'].strip().upper()

            # 去重
            dedup_key = (ts, price, volume, side)
            if dedup_key in seen:
                continue
            seen.add(dedup_key)

            # 剔除异常值
            if price <= 0 or volume <= 0:
                continue

            # 修正缺失方向
            if side not in ('B', 'S'):
                if last_price is None:
                    continue  # 首笔无法判断方向，丢弃
                if price > last_price:
                    side = 'B'
                elif price < last_price:
                    side = 'S'
                else:
                    side = last_side if last_side else 'B'
            if side not in ('B', 'S'):
                continue

            cleaned.append({
                'timestamp': ts,
                'price': price,
                'volume': volume,
                'side': side
            })
            last_price = price
            last_side = side

    # 写清洗后明细
    with open(output_detail, 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=['timestamp', 'price', 'volume', 'side'])
        writer.writeheader()
        writer.writerows(cleaned)

    # 按秒聚合
    features = {}
    for row in cleaned:
        dt = datetime.strptime(row['timestamp'], '%Y-%m-%d %H:%M:%S.%f')
        sec_key = dt.strftime('%Y-%m-%d %H:%M:%S')
        if sec_key not in features:
            features[sec_key] = {
                'total_volume': 0,
                'trade_count': 0,
                'open': row['price'],
                'high': row['price'],
                'low': row['price'],
                'close': row['price'],
                'buy_volume': 0,
                'sell_volume': 0
            }
        agg = features[sec_key]
        agg['total_volume'] += row['volume']
        agg['trade_count'] += 1
        agg['high'] = max(agg['high'], row['price'])
        agg['low'] = min(agg['low'], row['price'])
        agg['close'] = row['price']
        if row['side'] == 'B':
            agg['buy_volume'] += row['volume']
        else:
            agg['sell_volume'] += row['volume']

    # 计算价格变动（相对上一秒收盘）
    sorted_keys = sorted(features.keys())
    with open(output_feature, 'w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(['second', 'total_volume', 'trade_count', 'open', 'high', 'low', 'close',
                         'price_change', 'buy_volume', 'sell_volume'])
        prev_close = None
        for key in sorted_keys:
            agg = features[key]
            price_change = 0.0 if prev_close is None else round(agg['close'] - prev_close, 4)
            writer.writerow([key, agg['total_volume'], agg['trade_count'],
                             agg['open'], agg['high'], agg['low'], agg['close'],
                             price_change, agg['buy_volume'], agg['sell_volume']])
            prev_close = agg['close']

if __name__ == '__main__':
    clean_and_aggregate('trades_raw.csv', 'trades_clean.csv', 'trades_features.csv')
    print('处理完成，输出: trades_clean.csv, trades_features.csv')

运行与验证

准备输入文件trades_raw.csv（示例数据含重复行、零成交量、缺失方向记录），在命令行执行：

python tick_clean.py

验证方法：

打开trades_clean.csv，确认无重复行、无零成交量或非正价格记录，且每条side字段均为B或S。
打开trades_features.csv，确认每个second唯一，total_volume等于该秒内所有明细成交量之和，trade_count等于该秒明细条数，price_change等于当前秒close减上一秒close（首秒为0）。
用以下命令交叉校验聚合正确性：

python -c "
import csv
detail = list(csv.DictReader(open('trades_clean.csv')))
feat = list(csv.DictReader(open('trades_features.csv')))
sec = feat[0]['second']
vol = sum(int(r['volume']) for r in detail if r['timestamp'].startswith(sec))
cnt = sum(1 for r in detail if r['timestamp'].startswith(sec))
assert vol == int(feat[0]['total_volume']), 'volume mismatch'
assert cnt == int(feat[0]['trade_count']), 'count mismatch'
print('校验通过')
"

若输出校验通过，且清洗文件满足上述规则，则案例实现正确。