147 lines
6.8 KiB
Python
147 lines
6.8 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
from typing import Any
|
|
|
|
from backend.features.screener.catalog import (
|
|
ALLOWED_OPERATORS,
|
|
BUILTIN_STRATEGIES,
|
|
FACTOR_FIELDS,
|
|
REGIMES,
|
|
)
|
|
from backend.features.screener.indicators import _matches, _percentile_map, _risk_flags
|
|
|
|
|
|
class FormulaEvaluator:
|
|
def validate_formula(self, formula: dict[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(formula, dict):
|
|
raise ValueError("选股公式必须是 JSON 对象。")
|
|
result = copy.deepcopy(formula)
|
|
universe = result.setdefault("universe", {})
|
|
universe["exclude_st"] = bool(universe.get("exclude_st", True))
|
|
universe["listed_days_min"] = max(0, min(5000, int(universe.get("listed_days_min", 120))))
|
|
filters = result.setdefault("filters", [])
|
|
if not isinstance(filters, list) or len(filters) > 20:
|
|
raise ValueError("筛选条件必须是列表，且不能超过 20 条。")
|
|
for condition in filters:
|
|
field = condition.get("field")
|
|
operator = condition.get("op")
|
|
if field not in FACTOR_FIELDS:
|
|
raise ValueError(f"不支持的选股因子：{field}")
|
|
if operator not in ALLOWED_OPERATORS:
|
|
raise ValueError(f"不支持的运算符：{operator}")
|
|
if "value" not in condition:
|
|
raise ValueError(f"因子 {field} 缺少比较值。")
|
|
scores = result.setdefault("score", [])
|
|
if not isinstance(scores, list) or not scores or len(scores) > 12:
|
|
raise ValueError("评分因子应为 1 至 12 条。")
|
|
for item in scores:
|
|
if item.get("field") not in FACTOR_FIELDS:
|
|
raise ValueError(f"不支持的评分因子：{item.get('field')}")
|
|
item["weight"] = float(item.get("weight", 0))
|
|
if item["weight"] <= 0 or item["weight"] > 1:
|
|
raise ValueError("评分权重必须大于 0 且不超过 1。")
|
|
if item.get("direction", "desc") not in {"asc", "desc"}:
|
|
raise ValueError("评分方向只能是 asc 或 desc。")
|
|
item["direction"] = item.get("direction", "desc")
|
|
result["limit"] = max(1, min(50, int(result.get("limit", 15))))
|
|
result["min_score"] = max(0, min(1, float(result.get("min_score", 0))))
|
|
return result
|
|
|
|
def apply_formula(
|
|
self, rows: list[dict[str, Any]], formula: dict[str, Any], regime: str
|
|
) -> list[dict[str, Any]]:
|
|
universe = formula["universe"]
|
|
eligible = []
|
|
score_fields = [item["field"] for item in formula["score"]]
|
|
for row in rows:
|
|
name = str(row.get("name") or "")
|
|
if universe.get("exclude_st") and ("ST" in name.upper() or "退" in name):
|
|
continue
|
|
if row.get("listed_days", 0) < universe.get("listed_days_min", 0):
|
|
continue
|
|
if any(row.get(field) is None for field in score_fields):
|
|
continue
|
|
if all(_matches(row.get(item["field"]), item["op"], item["value"]) for item in formula["filters"]):
|
|
eligible.append(row)
|
|
if not eligible:
|
|
return []
|
|
|
|
percentiles = {
|
|
item["field"]: _percentile_map(eligible, item["field"], item["direction"])
|
|
for item in formula["score"]
|
|
}
|
|
weight_total = sum(item["weight"] for item in formula["score"])
|
|
results = []
|
|
for row in eligible:
|
|
contributions = []
|
|
score = 0.0
|
|
for item in formula["score"]:
|
|
percentile = percentiles[item["field"]].get(row["ts_code"], 0.5)
|
|
points = percentile * item["weight"] / weight_total
|
|
score += points
|
|
contributions.append(
|
|
{
|
|
"field": item["field"],
|
|
"label": FACTOR_FIELDS[item["field"]],
|
|
"value": row.get(item["field"], 0),
|
|
"points": round(points * 100, 1),
|
|
}
|
|
)
|
|
if score < formula["min_score"]:
|
|
continue
|
|
contributions.sort(key=lambda item: item["points"], reverse=True)
|
|
item = dict(row)
|
|
item["score"] = round(score, 4)
|
|
item["score_display"] = round(score * 100, 1)
|
|
item["contributions"] = contributions
|
|
item["reason"] = "、".join(entry["label"] for entry in contributions[:3])
|
|
include_regime_risk = formula.get("meta", {}).get("library") != "curated"
|
|
item["risk_flags"] = _risk_flags(row, regime, include_regime_risk)
|
|
results.append(item)
|
|
results.sort(key=lambda item: item["score"], reverse=True)
|
|
return results[: formula["limit"]]
|
|
|
|
|
|
def compile_local_strategy(prompt: str, regime: str) -> dict[str, Any]:
|
|
base = next((item for item in BUILTIN_STRATEGIES if regime in item["regimes"]), BUILTIN_STRATEGIES[1])
|
|
formula = copy.deepcopy(base["formula"])
|
|
description = prompt.strip() or base["description"]
|
|
lowered = description.lower()
|
|
if "低吸" in description:
|
|
formula["filters"] = [item for item in formula["filters"] if item["field"] != "pct_chg"]
|
|
formula["filters"].append({"field": "pct_chg", "op": "between", "value": [-3, 3]})
|
|
if "放量" in description:
|
|
formula["filters"].append({"field": "volume_ratio_5d", "op": ">=", "value": 1.2})
|
|
if "强势" in description or "突破" in description:
|
|
formula["filters"].append({"field": "return_5d", "op": ">=", "value": 5})
|
|
if "低波" in description or "稳健" in description:
|
|
formula["score"].append({"field": "volatility_10d", "weight": 0.18, "direction": "asc"})
|
|
if "资金" in description or "主力" in description:
|
|
formula["score"].append({"field": "net_flow_million", "weight": 0.18, "direction": "desc"})
|
|
if "小市值" in description or "小盘" in description:
|
|
formula["score"].append({"field": "circ_mv_billion", "weight": 0.15, "direction": "asc"})
|
|
if "竞价" in description:
|
|
formula["filters"].extend(
|
|
[
|
|
{"field": "auction_change", "op": "between", "value": [0.5, 8]},
|
|
{"field": "auction_amount_million", "op": ">=", "value": 2},
|
|
]
|
|
)
|
|
formula["score"].extend(
|
|
[
|
|
{"field": "auction_volume_ratio", "weight": 0.20, "direction": "desc"},
|
|
{"field": "auction_amount_million", "weight": 0.18, "direction": "desc"},
|
|
]
|
|
)
|
|
if "少量" in description or "精选" in description:
|
|
formula["limit"] = min(formula["limit"], 8)
|
|
formula["score"] = formula["score"][:12]
|
|
return {
|
|
"name": f"{REGIMES.get(regime, regime)}自定义策略",
|
|
"description": description,
|
|
"regimes": [regime],
|
|
"formula": formula,
|
|
"compiler": "local_template",
|
|
}
|