File size: 3,746 Bytes
590a501 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | """QuantaAlpha integration: factor library and signal building."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
import pandas as pd
from config.settings import PROJECT_ROOT
from data_pipeline.factor_loader import load_qlib_expression_factor
def load_library(path: str | Path) -> dict[str, Any]:
path = Path(path)
if not path.is_absolute():
path = PROJECT_ROOT / path
with open(path, encoding="utf-8") as f:
return json.load(f)
def list_factors(
library_path: str | Path,
quality_filter: str | None = None,
min_icir: float | None = None,
) -> pd.DataFrame:
data = load_library(library_path)
rows = []
for fid, info in data.get("factors", {}).items():
bt = info.get("backtest_results", {}) or {}
icir = bt.get("ICIR", bt.get("icir", bt.get("Rank ICIR")))
rows.append(
{
"factor_id": fid,
"factor_name": info.get("factor_name", fid),
"factor_expression": info.get("factor_expression", ""),
"factor_description": info.get("factor_description", ""),
"icir": icir,
"ic": bt.get("IC", bt.get("ic")),
"quality": info.get("quality", info.get("metadata", {}).get("quality")),
}
)
df = pd.DataFrame(rows)
if min_icir is not None and "icir" in df.columns:
df = df[df["icir"].fillna(-999) >= min_icir]
if quality_filter and "quality" in df.columns:
df = df[df["quality"].astype(str).str.lower() == quality_filter.lower()]
return df.sort_values("icir", ascending=False, na_position="last")
def build_signal_from_library(
library_path: str | Path,
factor_ids: list[str] | None = None,
top_k: int | None = None,
combine: str = "ic_weighted",
quality_filter: str | None = None,
min_icir: float | None = None,
start_time: str | None = None,
end_time: str | None = None,
) -> pd.Series:
"""Build combined signal from QuantaAlpha factor library JSON via qlib expressions."""
catalog = list_factors(library_path, quality_filter=quality_filter, min_icir=min_icir)
if factor_ids:
catalog = catalog[catalog["factor_id"].isin(factor_ids) | catalog["factor_name"].isin(factor_ids)]
if top_k:
catalog = catalog.head(top_k)
if catalog.empty:
raise ValueError("No factors selected from QuantaAlpha library")
series_list = []
weights = []
for _, row in catalog.iterrows():
expr = row["factor_expression"]
if not expr:
continue
name = row["factor_name"] or row["factor_id"]
s = load_qlib_expression_factor(
expression=expr,
start_time=start_time,
end_time=end_time,
name=name,
)
series_list.append(s.rename(name))
w = abs(float(row["icir"])) if pd.notna(row["icir"]) else 1.0
weights.append(w)
if not series_list:
raise ValueError("Selected QuantaAlpha factors have no qlib expressions")
mat = pd.concat(series_list, axis=1).sort_index()
if combine == "equal":
combined = mat.groupby(level="datetime").transform(lambda x: (x - x.mean()) / (x.std() + 1e-8)).mean(axis=1)
elif combine == "rank_mean":
combined = mat.groupby(level="datetime").rank(pct=True).mean(axis=1)
else:
w = pd.Series(weights, index=mat.columns)
w = w / w.sum()
normed = mat.groupby(level="datetime").transform(lambda x: (x - x.mean()) / (x.std() + 1e-8))
combined = normed.mul(w, axis=1).sum(axis=1)
combined.name = "score"
combined.index.names = ["instrument", "datetime"]
return combined
|