| """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 |
|
|
