# level2 **Repository Path**: airzou/level2 ## Basic Information - **Project Name**: level2 - **Description**: level2 - **Primary Language**: Unknown - **License**: Not specified - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 0 - **Created**: 2026-02-26 - **Last Updated**: 2026-02-28 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # level2-clean Level2 行情与逐笔数据清洗（uv 管理）。 **业务规则**：见 [规则.md](规则.md)（集合竞价 09:15–09:25、撤单、上交所/深交所 L2 差异等）。 ## 环境 ```bash # 安装 uv：https://docs.astral.sh/uv/ # 在项目根目录执行 uv sync ``` ## 结构与入口 经过重构，项目现在采用了标准的 Python 包结构，核心逻辑在 `src/` 下，执行入口在 `scripts/` 下。 ### 1. 核心 SDK (解析与清洗) 位于 **`src/level2_clean/`**，负责底层数据结构处理： - `io.py`（解析入口）：负责从本地目录或 `.7z` 压缩包中读取原始 CSV，并转化为 Pandas DataFrame。核心函数支持 `read_csv`、`open_7z_archive` 流式读取等。 - `clean.py`（清洗入口）：负责字段重命名、业务规则过滤、缺失值处理及时间类型转换。包含 `clean_hangqing`、`clean_zubi_weituo`、`clean_zubi_chengjiao` 三大核心函数，以及对齐 QuestDB 表结构的 `to_questdb_columns_xxx` 系列函数。 ### 2. 业务执行入口 所有的核心逻辑在上面两个文件中写好后，通过下面两个脚本进行统一调度。可以直接通过 `uv run python scripts/xxx.py` 运行： - **`scripts/process_daily_features.py`**（新特征计算入口）：支持多进程流式读取提取后的 CSV 或直接读 `.7z` 压缩包，在内存中跑通 L2 Analysis Pipelines，最后将降维后的核心指标存入 QuestDB 的 `l2_daily_features` 宽表。 - **示例 (直接读 7z 分析并入库)**： ```bash uv run python scripts/process_daily_features.py D:\BaiduNetdiskDownload\level2\data\20260205.7z 2026-02-05 ``` - **`scripts/ingest_from_dir.py`**（原数据流入口）：将解压后的 CSV 原始数据清洗后，全量明细灌入 QuestDB 对应的三张基础表中。 ### 3. 其他文件与辅助脚本 - `src/analysis/`：存放 P0~P6 的多步骤特征分析模型 (Pipeline)。 - `scripts/check_db.py`：检查 QuestDB 数据库连接状况与当日数据条数。 - `scripts/migrate_db.py`：用于初始化和更新数据库表结构。 - `scripts/inspect_7z.py`：查看并检测原始 7z 压缩包内的文件结构。 - `questdb_hangqing_create.sql`：QuestDB 三张基础表建表。 - `规则.md`：L2 业务规则说明（集合竞价、撤单、两所差异等）。 ## 使用 ### 推荐：先解压再处理（省时、无临时目录） 批量处理 7z 时建议**先整体解压一次**，再用本地路径读 CSV，避免每次读文件都解压、产生大量临时目录。 ```bash uv run python -c " from pathlib import Path from level2_clean.io import extract_7z_to_dir, read_csv, list_7z from level2_clean.clean import clean_hangqing # 一次解压到默认目录（如 20260205.7z -> ./20260205/） base_dir = extract_7z_to_dir(Path('20260205.7z')) # 按需列文件或直接 rglob for csv_path in base_dir.rglob('*.csv'): df = read_csv(csv_path, encoding='gbk') df = clean_hangqing(df) # 写入 QuestDB / Parquet 等 " ``` **注意**：解压会占用与解压后等量的磁盘空间，7z 较大时请保证目标盘空间充足。处理完后可由调用方删除解压目录以释放空间。 ### 按需从 7z 读（单归档单临时目录） 若不能先全量解压，可用 `open_7z_archive`：整个 with 内只建一个临时目录、只打开一次 7z，按需解压并读 CSV。 ```bash uv run python -c " from pathlib import Path from level2_clean.io import open_7z_archive from level2_clean.clean import clean_hangqing with open_7z_archive(Path('20260205.7z')) as ctx: df = ctx.read_csv_from_archive('20260205/000001.SZ/行情.csv', encoding='gbk') df = clean_hangqing(df) " ``` ### 其他用法 列 7z 内容、单次从 7z 读（每次会建临时目录，不推荐批量用）： ```bash uv run python -c " from pathlib import Path from level2_clean.io import read_csv_from_7z, list_7z from level2_clean.clean import clean_hangqing for n in list_7z(Path('20260205.7z'))[:20]: print(n) # 若已知路径，读并清洗行情（单次可接受，批量请用上面两种方式） # df = read_csv_from_7z(Path('20260205.7z'), '20260205/000001.SZ/行情.csv', encoding='gbk') # df = clean_hangqing(df) " ``` 读本地 CSV 并清洗： ```bash uv run python -c " from pathlib import Path from level2_clean.io import read_csv from level2_clean.clean import clean_hangqing df = read_csv(Path('行情.csv'), encoding='gbk') df = clean_hangqing(df) print(df.head()) " ``` ## 从解压目录写入 QuestDB 1. 确保已执行 `questdb_hangqing_create.sql` 建表，且 QuestDB 已启动。 2. 在项目根目录配置 `.env`：`QUESTDB_HOST`、`QUESTDB_PORT`（默认 8812）、`QUESTDB_USER`、`QUESTDB_PASSWORD`、`QUESTDB_DATABASE`（默认 qdb）。 3. 将某日 7z **解压到目录**后，在项目根目录执行（把路径换成你的解压目录）： ```bash uv run python scripts/ingest_from_dir.py D:\BaiduNetdiskDownload\20260206\20260206 ``` 脚本会递归查找该目录下所有 `行情.csv`、`逐笔委托.csv`、`逐笔成交.csv`，清洗后写入 QuestDB。 ## 后续 - 按标的/日期批量从 7z 读入并写 QuestDB 或 Parquet - 在 `clean.py` 里补具体列名与规则（时间格式、异常过滤等） sql SELECT * FROM l2_daily_features WHERE mean_vwap_skew < 0; CREATE TABLE IF NOT EXISTS l2_daily_features ( /* 1. 基础标识 */ ts TIMESTAMP, /* 交易日期 */ symbol SYMBOL, /* 股票代码 */ /* 2. P0: 宽表基座特征 (盘口迫切度) */ mean_rel_aggro DOUBLE, /* 平均主买/主卖侵略性 (0~1) */ median_inter_arrival_ms DOUBLE, /* 同方向订单到达毫秒级中位数耗时 (评判高频竞争度) */ /* 3. P1: 时间熵特征 (机器指令与算法伪装) */ algo_windows LONG, /* 被判定为机器/算法主导的 5 分钟时间窗数量 (低熵) */ total_windows LONG, /* 全天有效的 5 分钟时间窗总数 */ mean_algo_entropy DOUBLE, /* 算法单区域的平均香农时间熵 */ mean_retail_entropy DOUBLE, /* 散户单区域的平均香农时间熵 */ algo_total_amount DOUBLE, /* 算法(低熵)时段内促成的总成交金额 */ retail_total_amount DOUBLE, /* 散户(高熵)时段内促成的总成交金额 */ /* 4. P2: 对倒虚假交易特征 (自嗨水分) */ wash_trade_ratio DOUBLE, /* 全天虚假对倒交易占总成交的比例 */ wash_amount DOUBLE, /* 判定为虚假对倒成交的总金额 */ clean_amount DOUBLE, /* 剥离水分后的真实有效成交金额 */ /* 5. P3: 撤单欺诈特征 (Spoofing 操纵与试盘) */ spoof_count LONG, /* 发生严重撤单欺诈的时间窗总数 */ fake_pressure_count LONG, /* "假压单" (上方大额挂单后撤单，诱导卖出)的发生次数 */ fake_support_count LONG, /* "假托底" (下方大额挂单后撤单，诱导买入)的发生次数 */ mean_sell_otr DOUBLE, /* 卖方全天平均 OTR 撤单比 */ mean_buy_otr DOUBLE, /* 买方全天平均 OTR 撤单比 */ mean_cancel_distance_ticks DOUBLE, /* [新增] 均值撤单价格距离最优盘口的偏离跳数 (鉴别真假做市) */ /* 6. P4: 真实资金流与四象限行为 (走势定性) */ total_net_flow DOUBLE, /* 取消噪音后的全天真实中枢流向 */ q2_accumulation DOUBLE, /* 仅在 "Q2静默吸筹"(主力进+小阴跌) 时期的净买入量 */ mean_vwap_skew DOUBLE, /* VWAP 偏度 (正值表示当日量价重心在上方) */ q1_count LONG, /* 出现 "主动拉升" (流入+涨) 的时间窗数量 */ q2_count LONG, /* 出现 "静默吸筹" (流入+跌) 的时间窗数量 */ q3_count LONG, /* 出现 "主动出货" (流出+跌) 的时间窗数量 */ q4_count LONG, /* 出现 "拉高出货" (流出+涨) 的时间窗数量 */ /* 7. P5: 绝对主力大单与筹码结构特征 */ main_net_inflow DOUBLE, /* 超大单(>100万)净流入金额 */ retail_funds_net_inflow DOUBLE, /* 散户小单净流入金额 (反向参考) */ main_funds_buy_amount DOUBLE, /* 主动扫盘的超大单总买额 */ main_funds_sell_amount DOUBLE, /* 主动砸盘的超大单总卖额 */ mid_tier_net_inflow DOUBLE, /* [新增] 中单(20w~100w)净买入，防范 TWAP 拆单隐藏 */ trade_size_gini DOUBLE, /* [新增] 成交分配基尼系数 (0 极散，1 极度寡头包揽) */ /* 8. P6: 微观结构与流动性基座特征 (盘口深度) */ mean_spread DOUBLE, /* [新增] 全天平均相对买卖价差 Spread */ mean_obi_top5 DOUBLE, /* [新增] 前五档买压 vs 卖压失衡比 (OBI, 短期极强预测) */ amihud_illiquidity DOUBLE, /* [新增] Amihud流动性冲击因子 (全天均值，厚度反指标) */ /* 9. 数据管线日志与质量监控 */ total_records LONG, /* 原始逐笔成交总数 */ clean_records LONG, /* 清洗过滤后有效总数 */ large_order_records LONG /* 符合大单的发生次数 */ ) timestamp(ts) PARTITION BY DAY;