diff --git a/config/impact_chain_templates.yaml b/config/impact_chain_templates.yaml index 2027ca0..12e55d0 100644 --- a/config/impact_chain_templates.yaml +++ b/config/impact_chain_templates.yaml @@ -234,3 +234,38 @@ templates: sectors: ["消费", "汽车"] direction: bullish confidence_decay: 0.6 + + usd_strengthen: + event_pattern: "美元.*走强|美元.*走高|美元.*升值|美元指数.*突破|美元指数.*上涨|USD.*strengthen|dollar.*surge" + trigger_type: monetary + chain: + - order: 1 + impact: "黄金承压" + sectors: ["黄金", "贵金属"] + direction: bearish + confidence_decay: 0.9 + - order: 2 + impact: "人民币贬值压力" + sectors: ["航空", "造纸"] + direction: bearish + confidence_decay: 0.7 + - order: 3 + impact: "出口企业受益" + sectors: ["纺织服装", "家电", "电子"] + direction: bullish + confidence_decay: 0.6 + + usd_weaken: + event_pattern: "美元.*走弱|美元.*走低|美元.*贬值|美元指数.*下跌|美元指数.*回落|USD.*weaken|dollar.*fall" + trigger_type: monetary + chain: + - order: 1 + impact: "黄金上涨" + sectors: ["黄金", "贵金属"] + direction: bullish + confidence_decay: 0.9 + - order: 2 + impact: "人民币升值" + sectors: ["航空", "造纸"] + direction: bullish + confidence_decay: 0.7 diff --git a/src/intelligence/impact_chain.py b/src/intelligence/impact_chain.py index 8eec457..5bed87b 100644 --- a/src/intelligence/impact_chain.py +++ b/src/intelligence/impact_chain.py @@ -360,19 +360,79 @@ def get_stock_impact(self, symbol: str) -> str | None: } -class ImpactChainEngine: - """Constructs impact chains from macro events to affected assets. +def _build_sector_stock_map( + templates: dict[str, dict[str, Any]], +) -> dict[str, list[str]]: + """Invert the curated CHAIN_TEMPLATES into a ``sector -> [stocks]`` map. + + Engine B (``CausalChainConstructor``) resolves only *sectors* — + ``ImpactChainLink.affected_stocks`` is always empty. To preserve Engine A's + ``find_stock_impact`` contract we re-use the hand-authored stock bindings + that live in ``CHAIN_TEMPLATES``: each path binds ``affected_sectors`` to + ``affected_stocks``, so inverting gives a sector → stock lookup we can use to + re-populate the bridged chain's stocks. + """ + sector_stocks: dict[str, list[str]] = {} + for template in templates.values(): + for path in template.get("paths", []): + stocks = path.get("affected_stocks", []) + if not stocks: + continue + for sector in path.get("affected_sectors", []): + bucket = sector_stocks.setdefault(sector, []) + for stock in stocks: + if stock not in bucket: + bucket.append(stock) + return sector_stocks + + +# Map a chain link's order to a human-readable transmission lag. +_LAG_BY_ORDER: dict[int, str] = {0: "immediate", 1: "1-3d", 2: "1-2w"} - Phase 1: Template-based chain construction (rule engine). - Phase 2 (future): LLM-enhanced chain building with historical analogy search. + +def _magnitude_from_confidence(confidence: float) -> str: + """Bucket a link confidence into Engine A's magnitude vocabulary.""" + if confidence >= 0.7: + return "strong" + if confidence >= 0.4: + return "moderate" + return "weak" + + +class ImpactChainEngine: + """Adapter exposing Engine A's API on top of Engine B. + + This class used to own its own ``CHAIN_TEMPLATES`` rule base. It is now a + thin adapter over :class:`~src.intelligence.causal_chain.CausalChainConstructor` + (Engine B), which loads the canonical YAML templates + (``config/impact_chain_templates.yaml``) and matches events via regex. + + The public surface — ``detect_event_type``, ``build_chain``, + ``build_chains_for_event``, ``find_stock_impact`` and ``persist_chain`` — + and the ``ImpactChain`` / ``TransmissionPath`` dataclasses are preserved for + existing callers (relevance_scorer, rotation_engine, tool_registry, the web + API). + + Engine B resolves sectors but leaves ``affected_stocks`` empty, so this + adapter re-binds stocks using a ``sector -> [stocks]`` map inverted from the + legacy ``CHAIN_TEMPLATES`` (kept in this module solely as that data source). """ def __init__(self, config: dict[str, Any] | None = None) -> None: self._config = config or self._load_config() - self._templates = self._config.get("templates", CHAIN_TEMPLATES) - self._keywords = self._config.get("keywords", EVENT_KEYWORDS) + # CHAIN_TEMPLATES / EVENT_KEYWORDS are retained ONLY as the data source + # for the sector -> stock map used to defuse Engine B's empty + # affected_stocks. Chain construction itself is delegated to Engine B. + templates = self._config.get("templates", CHAIN_TEMPLATES) + self._sector_stocks = _build_sector_stock_map(templates) + + from src.intelligence.causal_chain import CausalChainConstructor + + self._constructor = CausalChainConstructor() logger.info( - "ImpactChainEngine initialized with %d templates", len(self._templates) + "ImpactChainEngine initialized (adapter over CausalChainConstructor); " + "%d sectors in stock map", + len(self._sector_stocks), ) @staticmethod @@ -382,25 +442,31 @@ def _load_config() -> dict[str, Any]: except FileNotFoundError: return {} + def _resolve_stocks(self, sectors: list[str]) -> list[str]: + """Resolve a list of sectors to stock codes via the inverted map.""" + stocks: list[str] = [] + for sector in sectors: + for stock in self._sector_stocks.get(sector, []): + if stock not in stocks: + stocks.append(stock) + return stocks + def detect_event_type(self, text: str) -> list[str]: - """Detect which event templates match the given text. + """Detect which Engine B template matches the given text. + + Delegates to ``CausalChainConstructor``'s regex template matching. Args: text: News headline, alert summary, or event description. Returns: - List of matching template keys, ordered by match confidence. + ``[template_name]`` if a template matches, else ``[]``. Engine B + returns a single best match (first matching regex), so this is at + most one element — a semantic change from the old multi-template, + keyword-count-ranked behaviour. """ - text_lower = text.lower() - matches: list[tuple[str, int]] = [] - - for template_key, keywords in self._keywords.items(): - hit_count = sum(1 for kw in keywords if kw.lower() in text_lower) - if hit_count > 0: - matches.append((template_key, hit_count)) - - matches.sort(key=lambda x: x[1], reverse=True) - return [m[0] for m in matches] + match = self._constructor._match_template(text) + return [match[0]] if match else [] def build_chain( self, @@ -409,53 +475,56 @@ def build_chain( ) -> ImpactChain | None: """Build an impact chain from an event description. + Delegates construction to Engine B and bridges the resulting + ``CausalChain`` into Engine A's ``ImpactChain`` shape, re-binding stocks + from the sector → stock map (Engine B leaves ``affected_stocks`` empty). + Args: event_text: The event description/headline. - template_key: Specific template to use; auto-detected if None. + template_key: Accepted for backward compatibility but ignored — + Engine B selects the template from the event text itself. Returns: - ImpactChain or None if no matching template. + ImpactChain or None if Engine B finds no matching template. """ - if template_key is None: - detected = self.detect_event_type(event_text) - if not detected: - logger.debug("No matching template for event: %s", event_text[:50]) - return None - template_key = detected[0] - - template = self._templates.get(template_key) - if template is None: - logger.warning("Template not found: %s", template_key) + causal = self._constructor.construct_chain({"title": event_text}) + if causal is None: + logger.debug("No matching template for event: %s", event_text[:50]) return None - paths = [ - TransmissionPath( - cause=p["cause"], - effect=p["effect"], - direction=p["direction"], - magnitude=p["magnitude"], - affected_sectors=p.get("affected_sectors", []), - affected_stocks=p.get("affected_stocks", []), - lag=p.get("lag", "immediate"), + paths: list[TransmissionPath] = [] + previous_impact = event_text + for link in causal.chain: + affected_stocks = link.affected_stocks or self._resolve_stocks(link.sectors) + paths.append( + TransmissionPath( + cause=previous_impact, + effect=link.impact, + direction="positive" if link.direction == "bullish" else "negative", + magnitude=_magnitude_from_confidence(link.confidence), + affected_sectors=link.sectors, + affected_stocks=affected_stocks, + lag=_LAG_BY_ORDER.get(link.order, "1-3m"), + ) ) - for p in template["paths"] - ] + previous_impact = link.impact chain = ImpactChain( chain_id=str(uuid.uuid4()), trigger_event=event_text, - trigger_type=template.get("trigger_type", "unknown"), + trigger_type=causal.event_type, timestamp=datetime.now(UTC), transmission_paths=paths, - confidence=0.7, # template-based = moderate confidence + confidence=causal.base_confidence, time_horizon="short_term", source="template", ) logger.info( - "Built impact chain for '%s' using template '%s': %d paths, %d sectors", + "Built impact chain for '%s' via Engine B template '%s': " + "%d paths, %d sectors", event_text[:40], - template_key, + causal.event_type, len(paths), len(chain.all_affected_sectors), ) @@ -466,18 +535,15 @@ def build_chain( return chain def build_chains_for_event(self, event_text: str) -> list[ImpactChain]: - """Build all matching impact chains for an event. + """Build matching impact chains for an event. - An event like "中东战争导致油价飙升" may match both - geopolitical_war and oil_surge templates. + Semantic shift: Engine A used to return one chain per matched keyword + template (a compound event like "中东战争导致油价飙升" could yield two + chains). Engine B returns a single best-match chain per event, so this + now returns ``[chain]`` (or ``[]`` if no template matches). """ - detected = self.detect_event_type(event_text) - chains = [] - for key in detected: - chain = self.build_chain(event_text, template_key=key) - if chain: - chains.append(chain) - return chains + chain = self.build_chain(event_text) + return [chain] if chain is not None else [] def persist_chain(self, chain: ImpactChain) -> None: """Persist an impact chain to SQLite for historical analysis.""" diff --git a/tests/unit/test_impact_chain.py b/tests/unit/test_impact_chain.py index ebfb2c5..8a35c32 100644 --- a/tests/unit/test_impact_chain.py +++ b/tests/unit/test_impact_chain.py @@ -1,8 +1,28 @@ -"""Tests for ImpactChainEngine.""" +"""Tests for ImpactChainEngine — now a thin adapter over CausalChainConstructor. + +The engine delegates chain construction to Engine B (``CausalChainConstructor``, +backed by ``config/impact_chain_templates.yaml``) and re-binds per-stock impacts +using a ``sector -> [stocks]`` map inverted from the legacy ``CHAIN_TEMPLATES``. +These tests assert the *adapter* behaviour, not the old in-code templates. + +Events below are chosen to actually match Engine B's YAML ``event_pattern`` +regexes (e.g. ``oil_surge``: "OPEC.*减产", ``fed_rate_cut``: "美联储.*降息"). +""" import pytest -from src.intelligence.impact_chain import ImpactChainEngine +from src.intelligence.impact_chain import ImpactChain, ImpactChainEngine + +# An event that matches Engine B's `oil_surge` template AND whose resolved +# sectors (石油/炼化/航运/物流/光伏/风电/新能源) overlap the legacy sector->stock +# map, so stocks get re-bound. This is the key to defusing "the trap". +OIL_EVENT = "OPEC意外减产推动油价飙升" +# 中国石化 — bound to the 石油/炼化 sectors in CHAIN_TEMPLATES; the oil_surge +# chain produces those sectors, so it must resolve via the sector->stock map. +OIL_STOCK = "600028" + +# An event matching Engine B's `fed_rate_cut` template. +FED_EVENT = "美联储宣布降息25个基点" @pytest.fixture @@ -11,98 +31,126 @@ def engine(): class TestEventDetection: - def test_war_keywords(self, engine): - matches = engine.detect_event_type("以色列与伊朗发生军事冲突") - assert "geopolitical_war" in matches + def test_detect_oil_event(self, engine): + """A known event returns its matched Engine B template name.""" + assert engine.detect_event_type(OIL_EVENT) == ["oil_surge"] - def test_oil_keywords(self, engine): - matches = engine.detect_event_type("原油价格突破100美元") - assert "oil_surge" in matches + def test_detect_fed_event(self, engine): + assert engine.detect_event_type(FED_EVENT) == ["fed_rate_cut"] - def test_usd_strengthen(self, engine): - matches = engine.detect_event_type("美元走强DXY突破105") - assert "usd_strengthen" in matches + def test_detect_returns_single_best_match(self, engine): + """Engine B returns one best match (semantic shift from old behaviour).""" + matches = engine.detect_event_type("美联储宣布降息且油价飙升") + assert len(matches) == 1 - def test_fed_hawkish(self, engine): - matches = engine.detect_event_type("美联储鹰派信号暗示加息") - assert "fed_hawkish" in matches - - def test_multi_match(self, engine): - matches = engine.detect_event_type("中东战争导致原油价格飙升") - assert len(matches) >= 2 - assert "geopolitical_war" in matches - assert "oil_surge" in matches - - def test_no_match(self, engine): - matches = engine.detect_event_type("今天天气很好") - assert matches == [] + def test_no_match_returns_empty(self, engine): + assert engine.detect_event_type("今天天气很好") == [] class TestChainBuilding: - def test_build_war_chain(self, engine): - chain = engine.build_chain("中东战争爆发", template_key="geopolitical_war") - assert chain is not None - assert chain.trigger_type == "geopolitical" - assert len(chain.transmission_paths) > 0 + def test_build_chain_returns_impact_chain(self, engine): + chain = engine.build_chain(OIL_EVENT) + assert isinstance(chain, ImpactChain) + assert chain.trigger_event == OIL_EVENT + assert chain.trigger_type == "oil_surge" assert chain.source == "template" - def test_build_usd_chain(self, engine): - chain = engine.build_chain("美元走强突破104", template_key="usd_strengthen") - assert chain is not None - # 002155 (湖南黄金) should be negatively affected - impact = chain.get_stock_impact("002155") - assert impact == "negative" + def test_build_chain_has_non_empty_paths(self, engine): + chain = engine.build_chain(OIL_EVENT) + assert len(chain.transmission_paths) > 0 - def test_auto_detect_and_build(self, engine): - chain = engine.build_chain("美联储宣布降息25个基点") + def test_build_chain_bridges_direction(self, engine): + """bullish -> positive, bearish -> negative.""" + chain = engine.build_chain(OIL_EVENT) + directions = {p.direction for p in chain.transmission_paths} + assert directions <= {"positive", "negative"} + # oil_surge has both bullish (炼化) and bearish (航运) links. + assert "positive" in directions + assert "negative" in directions + + def test_build_chain_lag_from_order(self, engine): + """First link (order 1) maps to the '1-3d' lag bucket.""" + chain = engine.build_chain(OIL_EVENT) + assert chain.transmission_paths[0].lag == "1-3d" + + def test_template_key_arg_is_ignored(self, engine): + """The legacy template_key arg is accepted but ignored (Engine B decides).""" + chain = engine.build_chain(OIL_EVENT, template_key="nonexistent") assert chain is not None - assert chain.trigger_type == "monetary" + assert chain.trigger_type == "oil_surge" def test_no_match_returns_none(self, engine): - chain = engine.build_chain("今天天气很好") - assert chain is None + assert engine.build_chain("今天天气很好") is None - def test_invalid_template_returns_none(self, engine): - chain = engine.build_chain("test", template_key="nonexistent") - assert chain is None + def test_build_chains_for_event_returns_single(self, engine): + chains = engine.build_chains_for_event(OIL_EVENT) + assert len(chains) == 1 + assert isinstance(chains[0], ImpactChain) - def test_all_affected_sectors(self, engine): - chain = engine.build_chain("战争", template_key="geopolitical_war") + def test_build_chains_for_event_empty_on_no_match(self, engine): + assert engine.build_chains_for_event("今天天气很好") == [] + + def test_affected_sectors_populated(self, engine): + chain = engine.build_chain(OIL_EVENT) sectors = chain.all_affected_sectors - assert "黄金" in sectors - assert "军工" in sectors + # From the oil_surge YAML template. + assert "石油" in sectors + assert "航运" in sectors + + +class TestStockResolution: + """The trap: Engine B leaves affected_stocks empty; the adapter re-binds + them from the sector->stock map. These tests prove the trap is defused.""" - def test_all_affected_stocks(self, engine): - chain = engine.build_chain("黄金大涨", template_key="gold_surge") - stocks = chain.all_affected_stocks - assert "002155" in stocks # 湖南黄金 + def test_stocks_resolved_from_sector_map(self, engine): + chain = engine.build_chain(OIL_EVENT) + # The oil chain's 石油/炼化 sectors must resolve to 中国石化 (600028). + assert OIL_STOCK in chain.all_affected_stocks + def test_find_stock_impact_non_empty(self, engine): + """find_stock_impact returns >=1 impact for a re-bound stock.""" + chains = engine.build_chains_for_event(OIL_EVENT) + impacts = engine.find_stock_impact(OIL_STOCK, chains) + assert len(impacts) >= 1 + # The 炼化 link is bullish -> positive for 中国石化. + assert any(i["direction"] == "positive" for i in impacts) -class TestMultipleChains: - def test_build_chains_for_compound_event(self, engine): - chains = engine.build_chains_for_event("中东战争导致原油价格飙升") - assert len(chains) >= 2 + def test_find_stock_impact_carries_chain_context(self, engine): + chains = engine.build_chains_for_event(OIL_EVENT) + impacts = engine.find_stock_impact(OIL_STOCK, chains) + first = impacts[0] + for key in ("chain_id", "trigger_event", "cause", "effect", "direction"): + assert key in first - def test_find_stock_impact(self, engine): - chains = engine.build_chains_for_event("美元走强黄金承压") - impacts = engine.find_stock_impact("002155", chains) - assert len(impacts) > 0 - assert any(i["direction"] == "negative" for i in impacts) + def test_find_stock_impact_empty_for_unknown_stock(self, engine): + chains = engine.build_chains_for_event(OIL_EVENT) + assert engine.find_stock_impact("999999", chains) == [] class TestSerialization: - def test_chain_to_dict(self, engine): - chain = engine.build_chain("战争", template_key="geopolitical_war") + def test_chain_to_dict_round_trips(self, engine): + chain = engine.build_chain(OIL_EVENT) d = chain.to_dict() - assert "chain_id" in d - assert "trigger_event" in d - assert "transmission_paths" in d - assert len(d["transmission_paths"]) > 0 - assert "cause" in d["transmission_paths"][0] + assert d["trigger_event"] == OIL_EVENT + assert d["trigger_type"] == "oil_surge" + assert d["source"] == "template" + assert len(d["transmission_paths"]) == len(chain.transmission_paths) + first_path = d["transmission_paths"][0] + for key in ( + "cause", + "effect", + "direction", + "magnitude", + "affected_sectors", + "affected_stocks", + "lag", + ): + assert key in first_path def test_path_to_dict(self, engine): - chain = engine.build_chain("战争", template_key="geopolitical_war") + chain = engine.build_chain(OIL_EVENT) path_dict = chain.transmission_paths[0].to_dict() assert "cause" in path_dict assert "direction" in path_dict assert "affected_sectors" in path_dict + assert "affected_stocks" in path_dict diff --git a/tests/unit/test_relevance_scorer.py b/tests/unit/test_relevance_scorer.py index 98b770d..bce6051 100644 --- a/tests/unit/test_relevance_scorer.py +++ b/tests/unit/test_relevance_scorer.py @@ -82,6 +82,13 @@ def test_war_triggers_gold_chain(self, scorer): assert any("影响链" in r for r in score.match_reasons) def test_usd_affects_gold(self, scorer): + """USD strength → gold transmission, preserved across the engine merge. + + The USD templates now live in the canonical YAML (config/ + impact_chain_templates.yaml: usd_strengthen / usd_weaken), and stocks + resolve through the shared sector→stock map (黄金 → 002155), so a bare + USD-strength headline still boosts a gold stock via the impact chain. + """ item = { "item_id": "i7", "title": "美元指数突破105", @@ -89,8 +96,8 @@ def test_usd_affects_gold(self, scorer): "related_symbols": [], } score = scorer.score(item, "002155", "湖南黄金") - # USD strengthening should trigger impact chain to gold assert score.relevance > 0 + assert any("影响链" in r for r in score.match_reasons) def test_oil_affects_petrochemical(self, scorer): item = {