394 lines
15 KiB
Python
394 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
data_comparator.py v1.0 — 同类数据横向对比与回测（v4.7 新增）
|
||
============================================================
|
||
对同类型的金融数据做：
|
||
1. 横向对比 — 多组同类序列（如多只股票的两融、多只 ETF 的资金流）逐字段对比，
|
||
输出最新值 / 均值 / 变化 / 排名。
|
||
2. 数据回测 — 对每组序列校验数据质量：完整性、时间连续性、数值异常跳变、
|
||
边界合理性，返回 0-100 质量评分与问题清单。
|
||
|
||
设计要点:
|
||
- 纯标准库，零外部依赖
|
||
- 泛化设计：接受任意 List[Dict] 序列，按数值字段自动识别
|
||
- 严格时间过滤由上游数据函数保证，本模块聚焦对比与质量校验
|
||
"""
|
||
|
||
import logging
|
||
from datetime import datetime
|
||
from typing import Dict, List, Any, Optional
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
# 默认数值字段识别：排除常见非数值/标识字段
|
||
_NON_NUMERIC_KEYS = {
|
||
"code", "name", "date", "market", "type", "title", "url", "person",
|
||
"position", "reason", "buyer_branch", "seller_branch", "warning",
|
||
"error", "update_time", "source", "curve_type", "id",
|
||
}
|
||
|
||
|
||
def _safe_float(val, default=0.0):
|
||
if val is None:
|
||
return default
|
||
if isinstance(val, str) and val.strip() in ("-", "--", "", "—"):
|
||
return default
|
||
try:
|
||
return float(val)
|
||
except (ValueError, TypeError):
|
||
return default
|
||
|
||
|
||
def _numeric_fields(records: List[Dict]) -> List[str]:
|
||
"""识别记录中的数值字段（排除标识类字段）。"""
|
||
if not records:
|
||
return []
|
||
fields = []
|
||
for key in records[0].keys():
|
||
if key.lower() in _NON_NUMERIC_KEYS:
|
||
continue
|
||
if any(key.lower().startswith(p) for p in ("_", "spread_")):
|
||
continue
|
||
# 取最多 30 条样本判断是否数值型
|
||
sample = [r.get(key) for r in records[:30]]
|
||
numeric = 0
|
||
for v in sample:
|
||
if v is None or (isinstance(v, str) and v.strip() in ("-", "", "--")):
|
||
continue
|
||
try:
|
||
float(v)
|
||
numeric += 1
|
||
except (ValueError, TypeError):
|
||
break
|
||
if numeric >= max(1, len(sample) // 2):
|
||
fields.append(key)
|
||
return fields
|
||
|
||
|
||
def _parse_date(d: str) -> Optional[str]:
|
||
"""规范化日期为 yyyy-MM-dd，失败返回 None。"""
|
||
if not d:
|
||
return None
|
||
s = str(d)[:10].strip()
|
||
for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d"):
|
||
try:
|
||
return datetime.strptime(s, fmt).strftime("%Y-%m-%d")
|
||
except ValueError:
|
||
continue
|
||
return None
|
||
|
||
|
||
# ==================== 数据回测 ====================
|
||
|
||
|
||
def backtest_series(label: str, records: List[Dict],
|
||
date_key: str = "date",
|
||
value_fields: Optional[List[str]] = None,
|
||
jump_threshold: float = 5.0) -> Dict[str, Any]:
|
||
"""对单个数据序列做质量回测。
|
||
|
||
校验维度:
|
||
1. 完整性 — 记录数量、日期覆盖率
|
||
2. 时间连续性 — 日期去重数、最大间隔（交易日/自然日）
|
||
3. 数值异常 — 环比跳变超过阈值的记录
|
||
4. 边界合理性 — 非正/缺失值占比
|
||
|
||
Args:
|
||
label: 序列名称（如 "600519 贵州茅台"）
|
||
records: 记录列表，每项含 date 与数值字段
|
||
date_key: 日期字段名
|
||
value_fields: 指定数值字段；为空则自动识别
|
||
jump_threshold: 环比跳变阈值（百分比），默认 500%（5.0 倍）
|
||
|
||
Returns:
|
||
{
|
||
"label": str, "record_count": int,
|
||
"score": int (0-100), "passed": bool,
|
||
"issues": [ {level, field, message}, ... ],
|
||
"by_dimension": {"completeness": {}, "continuity": {}, "numeric": {}, "boundary": {}},
|
||
"fields_checked": [...],
|
||
}
|
||
"""
|
||
issues: List[Dict[str, str]] = []
|
||
by_dim = {}
|
||
|
||
n = len(records)
|
||
if n == 0:
|
||
return {
|
||
"label": label, "record_count": 0, "score": 0, "passed": False,
|
||
"issues": [{"level": "error", "field": date_key, "message": "无任何记录"}],
|
||
"by_dimension": {}, "fields_checked": [],
|
||
}
|
||
|
||
fields = value_fields or _numeric_fields(records)
|
||
score = 100
|
||
dim = {}
|
||
|
||
# 1) 完整性
|
||
dates = [_parse_date(r.get(date_key, "")) for r in records]
|
||
dates = [d for d in dates if d]
|
||
unique_dates = len(set(dates))
|
||
completeness_issues = []
|
||
if unique_dates == 0:
|
||
completeness_issues.append({"level": "error", "field": date_key, "message": "缺少有效日期字段"})
|
||
score -= 30
|
||
elif unique_dates < max(2, n):
|
||
dup = n - unique_dates
|
||
if dup > 0:
|
||
completeness_issues.append({"level": "warn", "field": date_key, "message": f"存在 {dup} 条重复日期记录"})
|
||
score -= max(0, min(dup * 2, 15))
|
||
dim["completeness"] = {
|
||
"record_count": n, "unique_dates": unique_dates, "issue_count": len(completeness_issues),
|
||
}
|
||
issues.extend(completeness_issues)
|
||
|
||
# 2) 时间连续性（自然日间隔，跳过周末）
|
||
continuity_issues = []
|
||
if unique_dates >= 2:
|
||
sorted_dates = sorted(set(dates))
|
||
try:
|
||
gaps = []
|
||
for i in range(1, len(sorted_dates)):
|
||
d1 = datetime.strptime(sorted_dates[i - 1], "%Y-%m-%d")
|
||
d2 = datetime.strptime(sorted_dates[i], "%Y-%m-%d")
|
||
gaps.append((d2 - d1).days)
|
||
# 排除周末/节假日（1天正常；>4天视为可能缺失）
|
||
big_gaps = [g for g in gaps if g > 4]
|
||
max_gap = max(gaps)
|
||
if big_gaps:
|
||
continuity_issues.append({
|
||
"level": "warn", "field": date_key,
|
||
"message": f"存在 {len(big_gaps)} 处超长间隔（最大 {max_gap} 天，可能缺数据）",
|
||
})
|
||
score -= min(len(big_gaps) * 5, 20)
|
||
dim["continuity"] = {"span_days": (datetime.strptime(sorted_dates[-1], "%Y-%m-%d")
|
||
- datetime.strptime(sorted_dates[0], "%Y-%m-%d")).days,
|
||
"max_gap_days": max_gap, "big_gaps": len(big_gaps)}
|
||
except ValueError:
|
||
pass
|
||
issues.extend(continuity_issues)
|
||
|
||
# 3) 数值异常跳变
|
||
numeric_issues = []
|
||
for f in fields:
|
||
vals = [_safe_float(r.get(f)) for r in records]
|
||
# 只对正数值做环比跳变检测
|
||
for i in range(1, len(vals)):
|
||
prev, cur = vals[i - 1], vals[i]
|
||
if prev <= 0 or cur <= 0:
|
||
continue
|
||
ratio = abs(cur - prev) / abs(prev) * 100
|
||
if ratio > jump_threshold * 100:
|
||
numeric_issues.append({
|
||
"level": "warn", "field": f,
|
||
"message": f"第 {i + 1} 条相对上条跳变 {ratio:.1f}%（超 {jump_threshold * 100:.0f}%）",
|
||
})
|
||
score -= min(int(ratio // 100) * 3, 15)
|
||
break # 每字段最多报 1 条
|
||
dim["numeric"] = {"fields": fields, "issue_count": len(numeric_issues)}
|
||
issues.extend(numeric_issues)
|
||
|
||
# 4) 边界合理性
|
||
boundary_issues = []
|
||
for f in fields:
|
||
vals = [_safe_float(r.get(f)) for r in records]
|
||
non_positive = sum(1 for v in vals if v <= 0)
|
||
if non_positive > n * 0.5:
|
||
boundary_issues.append({
|
||
"level": "warn", "field": f,
|
||
"message": f"超过 50% 的记录该字段为 0 或负值（{non_positive}/{n}），疑似异常",
|
||
})
|
||
score -= 10
|
||
dim["boundary"] = {"issue_count": len(boundary_issues)}
|
||
issues.extend(boundary_issues)
|
||
|
||
# 汇总
|
||
by_dim = dim
|
||
score = max(0, min(100, score))
|
||
return {
|
||
"label": label,
|
||
"record_count": n,
|
||
"score": score,
|
||
"passed": score >= 60,
|
||
"issues": issues,
|
||
"by_dimension": by_dim,
|
||
"fields_checked": fields,
|
||
}
|
||
|
||
|
||
# ==================== 横向对比 ====================
|
||
|
||
|
||
def compare_series(series_dict: Dict[str, List[Dict]],
|
||
date_key: str = "date",
|
||
value_fields: Optional[List[str]] = None,
|
||
limit: int = 50) -> Dict[str, Any]:
|
||
"""横向对比多组同类数据序列。
|
||
|
||
Args:
|
||
series_dict: {"标签": [记录列表], ...}，如 {"贵州茅台": [...两融...], "五粮液": [...]}
|
||
date_key: 日期字段名
|
||
value_fields: 指定数值字段；为空则取各组共同数值字段的并集
|
||
limit: 每组取最近 N 条用于统计
|
||
|
||
Returns:
|
||
{
|
||
"series_count": int, "common_fields": [...],
|
||
"latest": {label: {field: 最新值}}, # 各序列最新值
|
||
"change": {label: {field: 环比变化}}, # 最新 vs 上一条
|
||
"ranking": {field: [ {label, value, rank}, ... ]}, # 按最新值排名
|
||
"per_series_stats": {label: {field: {latest, mean, min, max, change}}},
|
||
}
|
||
"""
|
||
if not series_dict:
|
||
return {"series_count": 0, "common_fields": [], "latest": {}, "change": {}, "ranking": {}, "per_series_stats": {}}
|
||
|
||
# 收集各组数值字段（并集）
|
||
all_fields: List[str] = []
|
||
for recs in series_dict.values():
|
||
for f in (_numeric_fields(recs) if value_fields is None else value_fields):
|
||
if f not in all_fields:
|
||
all_fields.append(f)
|
||
if value_fields:
|
||
all_fields = [f for f in all_fields if f in value_fields] or all_fields
|
||
|
||
latest: Dict[str, Dict[str, float]] = {}
|
||
change: Dict[str, Dict[str, float]] = {}
|
||
stats: Dict[str, Dict[str, Dict[str, float]]] = {}
|
||
|
||
for label, recs in series_dict.items():
|
||
recs = recs[-limit:] if limit else recs
|
||
latest[label] = {}
|
||
change[label] = {}
|
||
stats[label] = {}
|
||
for f in all_fields:
|
||
vals = [_safe_float(r.get(f)) for r in recs]
|
||
if not vals:
|
||
continue
|
||
latest_val = vals[-1]
|
||
prev_val = vals[-2] if len(vals) > 1 else 0.0
|
||
latest[label][f] = round(latest_val, 4)
|
||
change[label][f] = round(latest_val - prev_val, 4) if len(vals) > 1 else None
|
||
stats[label][f] = {
|
||
"latest": round(latest_val, 4),
|
||
"mean": round(sum(vals) / len(vals), 4),
|
||
"min": round(min(vals), 4),
|
||
"max": round(max(vals), 4),
|
||
"change": change[label][f],
|
||
}
|
||
|
||
# 排名（按最新值降序）
|
||
ranking: Dict[str, List[Dict[str, Any]]] = {}
|
||
for f in all_fields:
|
||
rows = []
|
||
for label, recs in series_dict.items():
|
||
vals = [_safe_float(r.get(f)) for r in (recs[-limit:] if limit else recs)]
|
||
if not vals:
|
||
continue
|
||
rows.append({"label": label, "value": round(vals[-1], 4)})
|
||
rows.sort(key=lambda x: x["value"], reverse=True)
|
||
for i, row in enumerate(rows, 1):
|
||
row["rank"] = i
|
||
ranking[f] = rows
|
||
|
||
return {
|
||
"series_count": len(series_dict),
|
||
"common_fields": all_fields,
|
||
"latest": latest,
|
||
"change": change,
|
||
"ranking": ranking,
|
||
"per_series_stats": stats,
|
||
}
|
||
|
||
|
||
def compare_and_backtest(series_dict: Dict[str, List[Dict]],
|
||
date_key: str = "date",
|
||
value_fields: Optional[List[str]] = None,
|
||
jump_threshold: float = 5.0,
|
||
limit: int = 50) -> Dict[str, Any]:
|
||
"""一键：横向对比 + 数据回测。
|
||
|
||
Args:
|
||
series_dict: {"标签": [记录列表], ...}
|
||
date_key: 日期字段名
|
||
value_fields: 指定数值字段
|
||
jump_threshold: 回测跳变阈值
|
||
limit: 每组参与统计的最近条数
|
||
|
||
Returns:
|
||
{
|
||
"comparison": {对比结果},
|
||
"backtests": [ {回测结果}, ... ],
|
||
"summary": {group_count, avg_score, best, worst},
|
||
}
|
||
"""
|
||
comparison = compare_series(series_dict, date_key, value_fields, limit)
|
||
backtests = [
|
||
backtest_series(label, recs, date_key, value_fields, jump_threshold)
|
||
for label, recs in series_dict.items()
|
||
]
|
||
|
||
scores = [b["score"] for b in backtests]
|
||
summary = {
|
||
"group_count": len(backtests),
|
||
"avg_score": round(sum(scores) / len(scores), 1) if scores else 0,
|
||
"best": max(backtests, key=lambda b: b["score"]) if backtests else None,
|
||
"worst": min(backtests, key=lambda b: b["score"]) if backtests else None,
|
||
}
|
||
# 精简 best/worst，避免嵌套过大
|
||
if summary.get("best"):
|
||
summary["best"] = {"label": summary["best"]["label"], "score": summary["best"]["score"]}
|
||
if summary.get("worst"):
|
||
summary["worst"] = {"label": summary["worst"]["label"], "score": summary["worst"]["score"]}
|
||
|
||
return {"comparison": comparison, "backtests": backtests, "summary": summary}
|
||
|
||
|
||
# ==================== 便捷函数 ====================
|
||
|
||
|
||
def format_comparison_report(result: Dict[str, Any]) -> str:
|
||
"""将对比回测结果格式化为可读文本。"""
|
||
comp = result.get("comparison", {})
|
||
lines = [f"📊 同类数据横向对比与回测（{comp.get('series_count', 0)} 组）", "=" * 50]
|
||
|
||
# 排名
|
||
for field, rows in comp.get("ranking", {}).items():
|
||
lines.append(f"\n【{field}】最新值排名:")
|
||
for row in rows[:5]:
|
||
lines.append(f" {row['rank']}. {row['label']}: {row['value']}")
|
||
|
||
# 回测评分
|
||
lines.append("\n【数据质量回测】")
|
||
for b in result.get("backtests", []):
|
||
status = "✅" if b["passed"] else "⚠️"
|
||
lines.append(f" {status} {b['label']}: {b['score']} 分（{b['record_count']} 条记录）")
|
||
for issue in b["issues"][:3]:
|
||
lines.append(f" - [{issue['level']}] {issue['message']}")
|
||
|
||
summary = result.get("summary", {})
|
||
lines.append(f"\n📈 汇总: 平均 {summary.get('avg_score', 0)} 分 | "
|
||
f"最佳 {summary.get('best', {}).get('label', '-')} | "
|
||
f"待关注 {summary.get('worst', {}).get('label', '-')}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ==================== CLI 测试入口 ====================
|
||
|
||
if __name__ == "__main__":
|
||
# 示例：两只股票的两融模拟数据
|
||
demo_a = [
|
||
{"date": "2026-07-28", "margin_balance": 100.0, "margin_buy": 5.0},
|
||
{"date": "2026-07-29", "margin_balance": 102.0, "margin_buy": 6.0},
|
||
{"date": "2026-07-30", "margin_balance": 105.0, "margin_buy": 4.0},
|
||
{"date": "2026-07-31", "margin_balance": 103.0, "margin_buy": 7.0},
|
||
]
|
||
demo_b = [
|
||
{"date": "2026-07-28", "margin_balance": 50.0, "margin_buy": 2.0},
|
||
{"date": "2026-07-29", "margin_balance": 53.0, "margin_buy": 3.0},
|
||
{"date": "2026-07-30", "margin_balance": 49.0, "margin_buy": 2.5},
|
||
{"date": "2026-07-31", "margin_balance": 52.0, "margin_buy": 4.0},
|
||
]
|
||
result = compare_and_backtest({"股票A": demo_a, "股票B": demo_b})
|
||
print(format_comparison_report(result))
|