File size: 4,546 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 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | """LLM-driven strategy selection and factor weighting (QuantaAlpha API integration)."""
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 integrations.quantaalpha.client import QuantaAlphaLLMClient, load_llm_config
from integrations.quantaalpha.factor_library import list_factors
STRATEGY_CATALOG = [
"topk_dropout",
"long_short_quantile",
"score_weighted_topk",
"rank_weighted",
"soft_topk",
"enhanced_indexing",
"dynamic_risk_topk",
"factor_equal_topk",
"factor_ic_weighted_topk",
]
def _build_factor_context(catalog: pd.DataFrame, max_factors: int = 30) -> str:
rows = catalog.head(max_factors).to_dict(orient="records")
return json.dumps(rows, ensure_ascii=False, indent=2)
def propose_strategy_with_llm(
factor_catalog: pd.DataFrame,
market_context: str | None = None,
client: QuantaAlphaLLMClient | None = None,
) -> dict[str, Any]:
"""
Ask LLM to propose strategy type, parameters, and factor selection.
Returns a dict compatible with strategies/registry.yaml entries.
"""
client = client or QuantaAlphaLLMClient(load_llm_config())
factor_json = _build_factor_context(factor_catalog)
system = (
"You are a quantitative portfolio strategist. "
"Given factor metadata, choose the best strategy from the catalog and parameters. "
f"Available strategies: {', '.join(STRATEGY_CATALOG)}. "
"Respond in JSON with keys: strategy_name, strategy_kwargs, selected_factor_ids, "
"signal_combine, rationale."
)
user = (
f"Market context: {market_context or 'CSI300 daily alpha strategy, out-of-sample backtest'}\n\n"
f"Factor catalog:\n{factor_json}\n\n"
"Pick 3-10 factors if using multi-factor combine strategies."
)
result = client.chat_json(
[
{"role": "system", "content": system},
{"role": "user", "content": user},
]
)
return result
def propose_from_quantaalpha_library(
library_path: str | Path,
market_context: str | None = None,
quality_filter: str | None = "high",
) -> dict[str, Any]:
catalog = list_factors(library_path, quality_filter=quality_filter)
if catalog.empty:
catalog = list_factors(library_path)
return propose_strategy_with_llm(catalog, market_context=market_context)
def build_llm_strategy_plan(
library_path: str | Path | None = None,
factor_panel_path: str | Path | None = None,
market_context: str | None = None,
) -> dict[str, Any]:
"""High-level entry: LLM plan -> signal source + strategy config."""
if library_path:
llm_plan = propose_from_quantaalpha_library(library_path, market_context=market_context)
signal_source = {
"type": "quantaalpha_library",
"path": str(library_path),
"factor_ids": llm_plan.get("selected_factor_ids"),
"combine": llm_plan.get("signal_combine", "ic_weighted"),
}
elif factor_panel_path:
from data_pipeline.factor_loader import load_factor_panel
panel = load_factor_panel(factor_panel_path)
factor_cols = [c for c in panel.columns if c.startswith("factor_")]
pseudo = pd.DataFrame({"factor_id": factor_cols, "factor_name": factor_cols, "icir": 1.0})
llm_plan = propose_strategy_with_llm(pseudo, market_context=market_context)
signal_source = {
"type": "factor_panel",
"path": str(factor_panel_path),
"factor_cols": llm_plan.get("selected_factor_ids") or factor_cols,
"combine": llm_plan.get("signal_combine", "equal"),
}
else:
raise ValueError("Provide library_path or factor_panel_path")
strategy_name = llm_plan.get("strategy_name", "topk_dropout")
if strategy_name not in STRATEGY_CATALOG:
strategy_name = "topk_dropout"
return {
"llm_plan": llm_plan,
"signal_source": signal_source,
"strategy": {
"name": strategy_name,
"kwargs": llm_plan.get("strategy_kwargs", {}),
},
}
def save_strategy_plan(plan: dict[str, Any], output_path: str | Path) -> Path:
path = Path(output_path)
if not path.is_absolute():
path = PROJECT_ROOT / path
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
json.dump(plan, f, ensure_ascii=False, indent=2)
return path
|