scripts/synthesize_edge_concepts.py
#!/usr/bin/env python3
"""Synthesize abstract edge concepts from detector tickets and hints."""
from __future__ import annotations
import argparse
import statistics
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import yaml
DEFAULT_EXPORTABLE_FAMILIES = {
"pivot_breakout",
"gap_up_continuation",
"panic_reversal",
"news_reaction",
}
HYPOTHESIS_TO_TITLE = {
"breakout": "Participation-backed trend breakout",
"earnings_drift": "Event-driven continuation drift",
"news_reaction": "Event overreaction and drift",
"futures_trigger": "Cross-asset propagation",
"calendar_anomaly": "Seasonality-linked demand imbalance",
"panic_reversal": "Shock overshoot mean reversion",
"regime_shift": "Regime transition opportunity",
"sector_x_stock": "Leader-laggard sector relay",
"research_hypothesis": "Unclassified edge hypothesis",
}
HYPOTHESIS_TO_THESIS = {
"breakout": (
"When liquidity and participation expand during a positive regime, "
"price expansion above structural pivots can persist for multiple sessions."
),
"earnings_drift": (
"Large information shocks can lead to underreaction, creating measurable post-event continuation."
),
"news_reaction": (
"Extreme single-day reactions often create either delayed continuation or overshoot reversion windows."
),
"futures_trigger": (
"Cross-asset futures shocks can transmit to related equities through hedging flows and risk transfer."
),
"calendar_anomaly": (
"Recurring calendar windows can produce repeatable demand-supply imbalances for specific symbols."
),
"panic_reversal": (
"Large downside shocks accompanied by exhaustion flow can set up short-horizon reversal edges."
),
"regime_shift": (
"Early inflections in breadth, correlation, and volatility can front-run major regime transitions."
),
"sector_x_stock": (
"Leadership shocks in one symbol can propagate into linked symbols through sector-level flow dynamics."
),
"research_hypothesis": (
"Observed pattern may represent a repeatable conditional edge requiring explicit validation."
),
}
HYPOTHESIS_TO_PLAYBOOKS = {
"breakout": ["trend_following_breakout", "confirmation_filtered_breakout"],
"earnings_drift": ["gap_continuation", "post_event_drift"],
"news_reaction": ["event_drift_continuation", "event_reversal"],
"futures_trigger": ["cross_asset_follow_through", "mapped_basket_rotation"],
"calendar_anomaly": ["seasonal_rotation", "seasonal_overlay"],
"panic_reversal": ["shock_reversal", "bounce_with_trend_filter"],
"regime_shift": ["regime_transition_probe"],
"sector_x_stock": ["leader_laggard_pair", "sector_relay_follow_through"],
"research_hypothesis": ["research_probe"],
}
HYPOTHESIS_TO_INVALIDATIONS = {
"breakout": [
"Breakout fails quickly with volume contraction.",
"Breadth weakens while correlations spike defensively.",
],
"earnings_drift": [
"Post-event day closes below event-day low.",
"Volume confirmation disappears after day 1-2.",
],
"news_reaction": [
"Reaction mean-reverts fully within 1-2 sessions.",
"No follow-through after confirmation filter.",
],
"futures_trigger": [
"Futures shock normalizes immediately.",
"Mapped equities show no directional sensitivity.",
],
"calendar_anomaly": [
"Recent years break the historical seasonal pattern.",
"Pattern only survives in illiquid tails.",
],
"panic_reversal": [
"Shock extends without stabilization signal.",
"Reversal only appears in low-liquidity outliers.",
],
"regime_shift": [
"Breadth and volatility revert to prior regime quickly.",
"Signal appears only during isolated macro events.",
],
"sector_x_stock": [
"Lead-lag correlation collapses out-of-sample.",
"Propagation depends on one-off events only.",
],
"research_hypothesis": [
"Out-of-sample behavior does not replicate.",
"Costs erase edge expectancy.",
],
}
KNOWN_HYPOTHESIS_TYPES: frozenset[str] = frozenset(
{
"breakout",
"earnings_drift",
"news_reaction",
"futures_trigger",
"calendar_anomaly",
"panic_reversal",
"regime_shift",
"sector_x_stock",
}
)
FALLBACK_HYPOTHESIS_TYPE = "research_hypothesis"
HYPOTHESIS_KEYWORDS: dict[str, list[str]] = {
"breakout": ["breakout", "pivot", "participation", "high20"],
"earnings_drift": ["earnings", "drift", "post-event", "post_event", "pead"],
"news_reaction": ["news", "reaction", "headline", "catalyst"],
"futures_trigger": ["futures", "cross-asset", "cross_asset", "propagation"],
"calendar_anomaly": ["calendar", "seasonal", "buyback", "blackout", "rebalance", "window"],
"panic_reversal": ["panic", "reversal", "shock", "overshoot", "mean-reversion", "bounce"],
"regime_shift": [
"regime",
"transition",
"inflection",
"shift",
"rotation",
"breadth divergence",
],
"sector_x_stock": ["sector", "leader", "laggard", "relay", "supply chain"],
}
SYNTHETIC_TICKET_PREFIX = "hint_promo_"
DEFAULT_SYNTHETIC_PRIORITY = 30.0
class ConceptSynthesisError(Exception):
"""Raised when concept synthesis fails."""
def infer_hypothesis_type(hint: dict[str, Any]) -> str:
"""Infer hypothesis_type from explicit field or keyword scan."""
explicit = hint.get("hypothesis_type")
if isinstance(explicit, str) and explicit.strip().lower() in KNOWN_HYPOTHESIS_TYPES:
return explicit.strip().lower()
text = (str(hint.get("title", "")) + " " + str(hint.get("observation", ""))).lower()
best_type: str | None = None
best_count = 0
for hyp_type, keywords in HYPOTHESIS_KEYWORDS.items():
count = sum(1 for kw in keywords if kw in text)
if count > best_count:
best_count = count
best_type = hyp_type
return best_type if best_type is not None else FALLBACK_HYPOTHESIS_TYPE
def promote_hints_to_tickets(
hints: list[dict[str, Any]],
synthetic_priority: float,
exportable_families: set[str] | None = None,
) -> list[dict[str, Any]]:
"""Promote qualifying hints to synthetic tickets."""
tickets: list[dict[str, Any]] = []
for idx, hint in enumerate(hints):
title = str(hint.get("title", "")).strip()
if not title:
continue
hypothesis = infer_hypothesis_type(hint)
mechanism = str(hint.get("mechanism_tag", "")).strip() or "uncertain"
regime = str(hint.get("regime_bias", "")).strip() or "Unknown"
families = (
exportable_families if exportable_families is not None else DEFAULT_EXPORTABLE_FAMILIES
)
entry_family_raw = hint.get("preferred_entry_family")
if isinstance(entry_family_raw, str) and entry_family_raw in families:
entry_family = entry_family_raw
else:
entry_family = "research_only"
sanitized = sanitize_identifier(title)
ticket_id = f"{SYNTHETIC_TICKET_PREFIX}{sanitized}_{idx}"
observation: dict[str, Any] = {}
symbols = hint.get("symbols", [])
if isinstance(symbols, list) and symbols:
first = str(symbols[0]).strip().upper()
if first:
observation["symbol"] = first
tickets.append(
{
"id": ticket_id,
"hypothesis_type": hypothesis,
"mechanism_tag": mechanism,
"regime": regime,
"entry_family": entry_family,
"priority_score": synthetic_priority,
"observation": observation,
"_synthetic": True,
}
)
return tickets
def cap_synthetic_tickets(
real_tickets: list[dict[str, Any]],
synthetic_tickets: list[dict[str, Any]],
max_ratio: float,
floor: int = 3,
) -> list[dict[str, Any]]:
"""Cap synthetic ticket count relative to real ticket count.
The allowed number of synthetic tickets is
``max(ceil(len(real_tickets) * max_ratio), floor)``.
When the synthetic list exceeds this limit the highest-priority entries
are kept.
"""
import math
limit = max(math.ceil(len(real_tickets) * max_ratio), floor)
if len(synthetic_tickets) <= limit:
return synthetic_tickets
# Keep highest priority first
ranked = sorted(synthetic_tickets, key=lambda t: t.get("priority_score", 0), reverse=True)
return ranked[:limit]
def sanitize_identifier(value: str) -> str:
"""Create a safe identifier from free text."""
lowered = "".join(ch.lower() if ch.isalnum() else "_" for ch in value)
compact = "_".join(part for part in lowered.split("_") if part)
return compact or "concept"
def safe_float(value: Any, default: float = 0.0) -> float:
"""Best-effort float conversion."""
try:
return float(value)
except (TypeError, ValueError):
return default
def read_hints(path: Path | None) -> list[dict[str, Any]]:
"""Read optional hints YAML."""
if path is None:
return []
payload = yaml.safe_load(path.read_text())
if payload is None:
return []
if isinstance(payload, list):
hints = payload
elif isinstance(payload, dict):
raw = payload.get("hints", [])
hints = raw if isinstance(raw, list) else []
else:
raise ConceptSynthesisError("hints file must be list or {hints: [...]} format")
return [hint for hint in hints if isinstance(hint, dict)]
def discover_ticket_files(tickets_dir: Path) -> list[Path]:
"""Discover ticket YAML files recursively."""
return sorted([path for path in tickets_dir.rglob("*.yaml") if path.is_file()])
def read_ticket(path: Path) -> dict[str, Any] | None:
"""Read one ticket YAML file."""
payload = yaml.safe_load(path.read_text())
if not isinstance(payload, dict):
return None
if "id" not in payload or "hypothesis_type" not in payload:
return None
return payload
def ticket_symbol(ticket: dict[str, Any]) -> str | None:
"""Extract representative symbol."""
observation = ticket.get("observation")
if isinstance(observation, dict):
symbol = observation.get("symbol")
if isinstance(symbol, str) and symbol.strip():
return symbol.strip().upper()
symbol = ticket.get("symbol")
if isinstance(symbol, str) and symbol.strip():
return symbol.strip().upper()
return None
def ticket_conditions(ticket: dict[str, Any]) -> list[str]:
"""Collect condition strings from ticket."""
conditions: list[str] = []
signal_definition = ticket.get("signal_definition")
if isinstance(signal_definition, dict):
raw = signal_definition.get("conditions")
if isinstance(raw, list):
for item in raw:
if isinstance(item, str) and item.strip():
conditions.append(item.strip())
entry = ticket.get("entry")
if isinstance(entry, dict):
raw = entry.get("conditions")
if isinstance(raw, list):
for item in raw:
if isinstance(item, str) and item.strip():
conditions.append(item.strip())
return conditions
def cluster_key(ticket: dict[str, Any]) -> tuple[str, str, str]:
"""Build clustering key."""
hypothesis = str(ticket.get("hypothesis_type", "unknown")).strip() or "unknown"
mechanism = str(ticket.get("mechanism_tag", "uncertain")).strip() or "uncertain"
regime = str(ticket.get("regime", "Unknown")).strip() or "Unknown"
return hypothesis, mechanism, regime
def choose_recommended_entry_family(
entry_counter: Counter[str],
exportable_families: set[str] | None = None,
) -> str | None:
"""Choose recommended exportable entry family from distribution."""
families = (
exportable_families if exportable_families is not None else DEFAULT_EXPORTABLE_FAMILIES
)
for family, _ in entry_counter.most_common():
if family in families:
return family
return None
def match_hint_titles(
hints: list[dict[str, Any]],
symbols: list[str],
regime: str,
recommended_entry_family: str | None,
) -> list[str]:
"""Match hints relevant to concept symbols/family/regime."""
symbol_set = set(symbols)
titles: list[str] = []
for hint in hints:
title = str(hint.get("title", "")).strip()
if not title:
continue
hint_symbols_raw = hint.get("symbols", [])
hint_symbols = {
str(symbol).strip().upper()
for symbol in hint_symbols_raw
if isinstance(symbol, str) and symbol.strip()
}
hint_regime = str(hint.get("regime_bias", "")).strip()
hint_family = hint.get("preferred_entry_family")
symbol_match = not hint_symbols or bool(symbol_set.intersection(hint_symbols))
regime_match = not hint_regime or hint_regime == regime
family_match = (
recommended_entry_family is None
or hint_family is None
or hint_family == recommended_entry_family
)
if symbol_match and regime_match and family_match:
titles.append(title)
return sorted(set(titles))[:10]
def build_concept(
key: tuple[str, str, str],
tickets: list[dict[str, Any]],
hints: list[dict[str, Any]],
exportable_families: set[str] | None = None,
) -> dict[str, Any]:
"""Build one concept payload from clustered tickets."""
hypothesis, mechanism, regime = key
priority_scores = [safe_float(ticket.get("priority_score")) for ticket in tickets]
avg_priority = statistics.mean(priority_scores) if priority_scores else 0.0
symbols = [symbol for ticket in tickets if (symbol := ticket_symbol(ticket)) is not None]
symbol_counter = Counter(symbols)
top_symbols = [symbol for symbol, _ in symbol_counter.most_common(10)]
entry_counter: Counter[str] = Counter()
condition_counter: Counter[str] = Counter()
ticket_ids: list[str] = []
synthetic_ticket_ids: list[str] = []
for ticket in tickets:
ticket_id = str(ticket.get("id", "")).strip()
is_synthetic = bool(ticket.get("_synthetic"))
if ticket_id:
if is_synthetic:
synthetic_ticket_ids.append(ticket_id)
else:
ticket_ids.append(ticket_id)
entry_family = ticket.get("entry_family")
if (
isinstance(entry_family, str)
and entry_family.strip()
and entry_family != "research_only"
and not is_synthetic
):
entry_counter[entry_family.strip()] += 1
for condition in ticket_conditions(ticket):
condition_counter[condition] += 1
families = (
exportable_families if exportable_families is not None else DEFAULT_EXPORTABLE_FAMILIES
)
recommended_entry_family = choose_recommended_entry_family(entry_counter, families)
export_ready_v1 = recommended_entry_family in families
concept_id = sanitize_identifier(f"edge_concept_{hypothesis}_{mechanism}_{regime}")
title = HYPOTHESIS_TO_TITLE.get(hypothesis, f"{hypothesis} concept")
thesis = HYPOTHESIS_TO_THESIS.get(
hypothesis,
"Observed pattern may represent a repeatable conditional edge requiring explicit validation.",
)
hint_titles = match_hint_titles(
hints=hints,
symbols=top_symbols,
regime=regime,
recommended_entry_family=recommended_entry_family,
)
has_synthetic = bool(synthetic_ticket_ids)
support_block: dict[str, Any] = {
"ticket_count": len(tickets),
"avg_priority_score": round(avg_priority, 2),
"symbols": top_symbols,
"entry_family_distribution": dict(entry_counter),
"representative_conditions": [
condition for condition, _ in condition_counter.most_common(6)
],
}
if has_synthetic:
support_block["real_ticket_count"] = len(ticket_ids)
support_block["synthetic_ticket_count"] = len(synthetic_ticket_ids)
evidence_block: dict[str, Any] = {
"ticket_ids": ticket_ids,
"matched_hint_titles": hint_titles,
}
if has_synthetic:
evidence_block["synthetic_ticket_ids"] = synthetic_ticket_ids
return {
"id": concept_id,
"title": title,
"hypothesis_type": hypothesis,
"mechanism_tag": mechanism,
"regime": regime,
"support": support_block,
"abstraction": {
"thesis": thesis,
"invalidation_signals": HYPOTHESIS_TO_INVALIDATIONS.get(
hypothesis,
["Out-of-sample behavior does not replicate.", "Costs erase edge expectancy."],
),
},
"strategy_design": {
"playbooks": HYPOTHESIS_TO_PLAYBOOKS.get(hypothesis, ["research_probe"]),
"recommended_entry_family": recommended_entry_family,
"export_ready_v1": bool(export_ready_v1),
},
"evidence": evidence_block,
}
# ---------------------------------------------------------------------------
# Concept deduplication
# ---------------------------------------------------------------------------
def condition_overlap_ratio(conds_a: list[str], conds_b: list[str]) -> float:
"""Compute containment overlap: |A ∩ B| / min(|A|, |B|).
Returns 0.0 when either list is empty.
"""
if not conds_a or not conds_b:
return 0.0
set_a = {c.strip().lower() for c in conds_a}
set_b = {c.strip().lower() for c in conds_b}
intersection = len(set_a & set_b)
return intersection / min(len(set_a), len(set_b))
def merge_concepts(
primary: dict[str, Any],
secondary: dict[str, Any],
exportable_families: set[str] | None = None,
) -> dict[str, Any]:
"""Merge two concepts. The one with more tickets becomes *primary*.
Returns a new concept dict with combined support and evidence.
"""
p_count = primary.get("support", {}).get("ticket_count", 0)
s_count = secondary.get("support", {}).get("ticket_count", 0)
if s_count > p_count:
primary, secondary = secondary, primary
p_count, s_count = s_count, p_count
total_count = p_count + s_count
p_priority = primary.get("support", {}).get("avg_priority_score", 0.0)
s_priority = secondary.get("support", {}).get("avg_priority_score", 0.0)
avg_priority = round(
(p_priority * p_count + s_priority * s_count) / total_count if total_count else 0.0,
2,
)
# Symbols: union preserving order
p_symbols = list(primary.get("support", {}).get("symbols", []))
s_symbols = secondary.get("support", {}).get("symbols", [])
seen = set(p_symbols)
for sym in s_symbols:
if sym not in seen:
p_symbols.append(sym)
seen.add(sym)
# Mechanism tag
p_mech = str(primary.get("mechanism_tag", "uncertain"))
s_mech = str(secondary.get("mechanism_tag", "uncertain"))
if p_mech != s_mech:
merged_mechanism = "+".join(sorted([p_mech, s_mech]))
else:
merged_mechanism = p_mech
# Entry family distribution (Counter merge)
p_dist = dict(primary.get("support", {}).get("entry_family_distribution", {}))
for k, v in secondary.get("support", {}).get("entry_family_distribution", {}).items():
p_dist[k] = p_dist.get(k, 0) + v
# Representative conditions: union, cap at 6
p_conds = list(primary.get("support", {}).get("representative_conditions", []))
s_conds = secondary.get("support", {}).get("representative_conditions", [])
conds_lower_seen = {c.strip().lower() for c in p_conds}
for c in s_conds:
if c.strip().lower() not in conds_lower_seen:
p_conds.append(c)
conds_lower_seen.add(c.strip().lower())
p_conds = p_conds[:6]
# Entry family: primary preferred, fallback to secondary
p_family = primary.get("strategy_design", {}).get("recommended_entry_family")
s_family = secondary.get("strategy_design", {}).get("recommended_entry_family")
families = (
exportable_families if exportable_families is not None else DEFAULT_EXPORTABLE_FAMILIES
)
recommended = p_family if p_family is not None else s_family
export_ready = recommended in families if recommended else False
# Evidence: ticket_ids union
p_ticket_ids = list(primary.get("evidence", {}).get("ticket_ids", []))
s_ticket_ids = secondary.get("evidence", {}).get("ticket_ids", [])
all_ticket_ids = p_ticket_ids + [t for t in s_ticket_ids if t not in set(p_ticket_ids)]
# Evidence: hint titles union
p_hints = set(primary.get("evidence", {}).get("matched_hint_titles", []))
s_hints = set(secondary.get("evidence", {}).get("matched_hint_titles", []))
all_hints = sorted(p_hints | s_hints)
# ID regeneration
hypothesis = str(primary.get("hypothesis_type", "unknown"))
regime = str(primary.get("regime", "Unknown"))
concept_id = sanitize_identifier(f"edge_concept_{hypothesis}_{merged_mechanism}_{regime}")
# Build support block
support_block: dict[str, Any] = {
"ticket_count": total_count,
"avg_priority_score": avg_priority,
"symbols": p_symbols,
"entry_family_distribution": p_dist,
"representative_conditions": p_conds,
}
# Synthetic fields
p_real = primary.get("support", {}).get("real_ticket_count")
s_real = secondary.get("support", {}).get("real_ticket_count")
p_synth = primary.get("support", {}).get("synthetic_ticket_count")
s_synth = secondary.get("support", {}).get("synthetic_ticket_count")
if p_real is not None or s_real is not None:
support_block["real_ticket_count"] = (p_real or 0) + (s_real or 0)
if p_synth is not None or s_synth is not None:
support_block["synthetic_ticket_count"] = (p_synth or 0) + (s_synth or 0)
evidence_block: dict[str, Any] = {
"ticket_ids": all_ticket_ids,
"matched_hint_titles": all_hints,
}
p_synth_ids = primary.get("evidence", {}).get("synthetic_ticket_ids", [])
s_synth_ids = secondary.get("evidence", {}).get("synthetic_ticket_ids", [])
if p_synth_ids or s_synth_ids:
evidence_block["synthetic_ticket_ids"] = list(p_synth_ids) + [
t for t in s_synth_ids if t not in set(p_synth_ids)
]
merged = {
"id": concept_id,
"title": primary.get("title", ""),
"hypothesis_type": hypothesis,
"mechanism_tag": merged_mechanism,
"regime": regime,
"support": support_block,
"abstraction": dict(primary.get("abstraction", {})),
"strategy_design": {
"playbooks": list(primary.get("strategy_design", {}).get("playbooks", [])),
"recommended_entry_family": recommended,
"export_ready_v1": bool(export_ready),
},
"evidence": evidence_block,
"merged_from": [secondary.get("id", "unknown")],
}
return merged
def deduplicate_concepts(
concepts: list[dict[str, Any]],
overlap_threshold: float = 0.75,
exportable_families: set[str] | None = None,
) -> tuple[list[dict[str, Any]], int]:
"""Greedy pairwise deduplication within same hypothesis_type.
Returns ``(deduplicated_list, merge_count)``.
"""
if len(concepts) <= 1:
return list(concepts), 0
merged_indices: set[int] = set()
result: list[dict[str, Any]] = []
merge_count = 0
for i in range(len(concepts)):
if i in merged_indices:
continue
current = concepts[i]
for j in range(i + 1, len(concepts)):
if j in merged_indices:
continue
candidate = concepts[j]
if current.get("hypothesis_type") != candidate.get("hypothesis_type"):
continue
conds_a = current.get("support", {}).get("representative_conditions", [])
conds_b = candidate.get("support", {}).get("representative_conditions", [])
if condition_overlap_ratio(conds_a, conds_b) > overlap_threshold:
current = merge_concepts(current, candidate, exportable_families)
merged_indices.add(j)
merge_count += 1
result.append(current)
return result, merge_count
def parse_args() -> argparse.Namespace:
"""Parse CLI args."""
parser = argparse.ArgumentParser(
description="Synthesize abstract edge concepts from detector tickets.",
)
parser.add_argument(
"--tickets-dir", required=True, help="Directory containing ticket YAML files"
)
parser.add_argument("--hints", default=None, help="Optional hints YAML path")
parser.add_argument(
"--output",
default="reports/edge_concepts/edge_concepts.yaml",
help="Output concept YAML path",
)
parser.add_argument(
"--min-ticket-support",
type=int,
default=1,
help="Minimum ticket count required to keep a concept",
)
parser.add_argument(
"--promote-hints",
action="store_true",
default=False,
help="Promote qualifying hints to synthetic tickets for concept creation",
)
parser.add_argument(
"--synthetic-priority",
type=float,
default=DEFAULT_SYNTHETIC_PRIORITY,
help="Priority score for synthetic tickets (default: 30.0)",
)
parser.add_argument(
"--max-synthetic-ratio",
type=float,
default=None,
help="Max synthetic/real ticket ratio (e.g. 1.5); uncapped when omitted",
)
parser.add_argument(
"--overlap-threshold",
type=float,
default=0.75,
help="Condition overlap threshold for concept deduplication (default: 0.75)",
)
parser.add_argument(
"--no-dedup",
action="store_true",
default=False,
help="Disable concept deduplication",
)
parser.add_argument(
"--exportable-families",
default=None,
help="Comma-separated list of exportable entry families (overrides module default)",
)
return parser.parse_args()
def main() -> int:
"""CLI entrypoint."""
args = parse_args()
tickets_dir = Path(args.tickets_dir).resolve()
hints_path = Path(args.hints).resolve() if args.hints else None
output_path = Path(args.output).resolve()
ef_override: set[str] | None = None
if args.exportable_families:
ef_override = {f.strip() for f in args.exportable_families.split(",") if f.strip()}
if not tickets_dir.exists():
print(f"[ERROR] tickets dir not found: {tickets_dir}")
return 1
if hints_path is not None and not hints_path.exists():
print(f"[ERROR] hints file not found: {hints_path}")
return 1
try:
hints = read_hints(hints_path)
ticket_files = discover_ticket_files(tickets_dir)
tickets: list[dict[str, Any]] = []
for ticket_file in ticket_files:
ticket = read_ticket(ticket_file)
if ticket is not None:
tickets.append(ticket)
synthetic_tickets: list[dict[str, Any]] = []
if args.promote_hints and hints:
synthetic_tickets = promote_hints_to_tickets(
hints=hints,
synthetic_priority=args.synthetic_priority,
exportable_families=ef_override,
)
if args.max_synthetic_ratio is not None:
synthetic_tickets = cap_synthetic_tickets(
real_tickets=tickets,
synthetic_tickets=synthetic_tickets,
max_ratio=args.max_synthetic_ratio,
)
tickets = tickets + synthetic_tickets
if not tickets:
raise ConceptSynthesisError("no valid ticket files found")
grouped: dict[tuple[str, str, str], list[dict[str, Any]]] = defaultdict(list)
for ticket in tickets:
grouped[cluster_key(ticket)].append(ticket)
concepts: list[dict[str, Any]] = []
for key, cluster_tickets in grouped.items():
if len(cluster_tickets) < max(args.min_ticket_support, 1):
continue
concepts.append(
build_concept(
key=key, tickets=cluster_tickets, hints=hints, exportable_families=ef_override
)
)
# Deduplication
if not args.no_dedup:
concepts, dedup_merged_count = deduplicate_concepts(
concepts, args.overlap_threshold, ef_override
)
else:
dedup_merged_count = 0
concepts.sort(
key=lambda item: (
safe_float(item.get("support", {}).get("avg_priority_score")),
safe_float(item.get("support", {}).get("ticket_count")),
),
reverse=True,
)
if not concepts:
raise ConceptSynthesisError("no concepts passed min-ticket-support filter")
candidate_dates = [str(ticket.get("date")) for ticket in tickets if ticket.get("date")]
as_of = max(candidate_dates) if candidate_dates else None
source_block: dict[str, Any] = {
"tickets_dir": str(tickets_dir),
"hints_path": str(hints_path) if hints_path else None,
"ticket_file_count": len(ticket_files),
"ticket_count": len(tickets),
}
if args.promote_hints:
source_block["promote_hints"] = True
source_block["real_ticket_count"] = len(tickets) - len(synthetic_tickets)
source_block["synthetic_ticket_count"] = len(synthetic_tickets)
source_block["dedup_enabled"] = not args.no_dedup
source_block["overlap_threshold"] = args.overlap_threshold
source_block["dedup_merged_count"] = dedup_merged_count
payload = {
"generated_at_utc": datetime.now(timezone.utc).replace(microsecond=0).isoformat(),
"as_of": as_of,
"source": source_block,
"concept_count": len(concepts),
"concepts": concepts,
}
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(yaml.safe_dump(payload, sort_keys=False), encoding="utf-8")
except ConceptSynthesisError as exc:
print(f"[ERROR] {exc}")
return 1
synth_msg = f" synthetic_tickets={len(synthetic_tickets)}" if synthetic_tickets else ""
print(f"[OK] concepts={len(concepts)}{synth_msg} output={output_path}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/tests/test_synthesize_edge_concepts.py
"""Unit tests for synthesize_edge_concepts.py."""
from __future__ import annotations
from pathlib import Path
import synthesize_edge_concepts as sec
import yaml
def test_build_concept_for_breakout_is_export_ready() -> None:
tickets = [
{
"id": "edge_auto_vcp_xp_20260220",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"priority_score": 74.2,
"entry_family": "pivot_breakout",
"observation": {"symbol": "XP"},
"signal_definition": {"conditions": ["close > high20_prev", "rel_volume >= 1.5"]},
},
{
"id": "edge_auto_vcp_nok_20260220",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"priority_score": 73.0,
"entry_family": "pivot_breakout",
"observation": {"symbol": "NOK"},
"signal_definition": {"conditions": ["close > high20_prev"]},
},
]
concept = sec.build_concept(
key=("breakout", "behavior", "RiskOn"),
tickets=tickets,
hints=[
{
"title": "Breadth-supported breakout regime",
"preferred_entry_family": "pivot_breakout",
"regime_bias": "RiskOn",
}
],
)
assert concept["strategy_design"]["export_ready_v1"] is True
assert concept["strategy_design"]["recommended_entry_family"] == "pivot_breakout"
assert concept["support"]["ticket_count"] == 2
def test_infer_hypothesis_type_from_explicit_field() -> None:
"""Test explicit hypothesis_type within whitelist is accepted."""
hint = {"hypothesis_type": "breakout", "title": "whatever"}
assert sec.infer_hypothesis_type(hint) == "breakout"
def test_infer_hypothesis_type_case_insensitive() -> None:
"""Test that explicit field is case-insensitive."""
assert sec.infer_hypothesis_type({"hypothesis_type": "Breakout"}) == "breakout"
assert sec.infer_hypothesis_type({"hypothesis_type": " PANIC_REVERSAL "}) == "panic_reversal"
def test_infer_hypothesis_type_rejects_unknown_explicit() -> None:
"""Test unknown explicit value falls back to keyword inference."""
# "momentum" is not in whitelist, but title has "breakout" keyword
hint = {"hypothesis_type": "momentum", "title": "breakout pattern detected"}
assert sec.infer_hypothesis_type(hint) == "breakout"
# unknown explicit, no keyword match at all
hint2 = {"hypothesis_type": "xyz_typo", "title": "no match here"}
assert sec.infer_hypothesis_type(hint2) == sec.FALLBACK_HYPOTHESIS_TYPE
def test_infer_hypothesis_type_from_keywords() -> None:
"""Test keyword-based inference from title and observation."""
hint = {"title": "Seasonal buyback blackout window", "observation": "Calendar effect detected"}
assert sec.infer_hypothesis_type(hint) == "calendar_anomaly"
def test_infer_hypothesis_type_fallback() -> None:
"""Test fallback to FALLBACK_HYPOTHESIS_TYPE when no match."""
hint = {"title": "Unclear signal", "observation": "Unknown pattern"}
assert sec.infer_hypothesis_type(hint) == sec.FALLBACK_HYPOTHESIS_TYPE
def test_promote_hints_to_tickets_basic() -> None:
"""Test basic hint-to-ticket promotion."""
hints = [
{
"title": "March buyback blackout",
"observation": "Buyback blackout period",
"hypothesis_type": "calendar_anomaly",
"preferred_entry_family": "pivot_breakout",
"symbols": ["SPY"],
"regime_bias": "Neutral",
"mechanism_tag": "flow",
}
]
tickets = sec.promote_hints_to_tickets(hints, synthetic_priority=30.0)
assert len(tickets) == 1
t = tickets[0]
assert t["id"].startswith(sec.SYNTHETIC_TICKET_PREFIX)
assert t["hypothesis_type"] == "calendar_anomaly"
assert t["priority_score"] == 30.0
assert t["_synthetic"] is True
assert t["entry_family"] == "pivot_breakout"
assert t["observation"]["symbol"] == "SPY"
def test_promote_hints_to_tickets_skips_empty_title() -> None:
"""Test that hints with empty title are skipped."""
hints = [
{"title": "", "observation": "No title"},
{"title": " ", "observation": "Whitespace only"},
{"title": "Valid", "observation": "ok"},
]
tickets = sec.promote_hints_to_tickets(hints, synthetic_priority=30.0)
assert len(tickets) == 1
assert "valid" in tickets[0]["id"]
def test_promote_hints_to_tickets_defaults() -> None:
"""Test default values for minimal hint."""
hints = [{"title": "Minimal hint"}]
tickets = sec.promote_hints_to_tickets(hints, synthetic_priority=25.0)
assert len(tickets) == 1
t = tickets[0]
assert t["mechanism_tag"] == "uncertain"
assert t["regime"] == "Unknown"
assert t["entry_family"] == "research_only"
assert t["priority_score"] == 25.0
def test_build_concept_with_synthetic_tickets() -> None:
"""Test that build_concept separates real and synthetic ticket counts."""
tickets = [
{
"id": "edge_auto_real_1",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"priority_score": 74.0,
"entry_family": "pivot_breakout",
"observation": {"symbol": "XP"},
},
{
"id": "hint_promo_seasonal_0",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"priority_score": 30.0,
"entry_family": "research_only",
"_synthetic": True,
},
]
concept = sec.build_concept(
key=("breakout", "behavior", "RiskOn"),
tickets=tickets,
hints=[],
)
assert concept["support"]["ticket_count"] == 2
assert concept["support"]["real_ticket_count"] == 1
assert concept["support"]["synthetic_ticket_count"] == 1
assert concept["evidence"]["synthetic_ticket_ids"] == ["hint_promo_seasonal_0"]
def test_build_concept_synthetic_only_not_export_ready() -> None:
"""Test that hint-only concepts are NOT export_ready even with exportable entry_family."""
tickets = [
{
"id": "hint_promo_breakout_0",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"priority_score": 30.0,
"entry_family": "pivot_breakout",
"_synthetic": True,
},
]
concept = sec.build_concept(
key=("breakout", "behavior", "RiskOn"),
tickets=tickets,
hints=[],
)
assert concept["strategy_design"]["export_ready_v1"] is False
assert concept["strategy_design"]["recommended_entry_family"] is None
assert concept["support"]["entry_family_distribution"] == {}
def test_build_concept_without_synthetic_omits_breakdown() -> None:
"""Test backward compatibility: no synthetic fields when all real."""
tickets = [
{
"id": "edge_auto_real_1",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"priority_score": 74.0,
"entry_family": "pivot_breakout",
"observation": {"symbol": "XP"},
},
]
concept = sec.build_concept(
key=("breakout", "behavior", "RiskOn"),
tickets=tickets,
hints=[],
)
assert "real_ticket_count" not in concept["support"]
assert "synthetic_ticket_count" not in concept["support"]
assert "synthetic_ticket_ids" not in concept["evidence"]
def test_build_concept_for_news_reaction_is_research_only() -> None:
concept = sec.build_concept(
key=("news_reaction", "behavior", "RiskOn"),
tickets=[
{
"id": "edge_auto_news_reaction_tsla_20260220",
"hypothesis_type": "news_reaction",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"priority_score": 90.0,
"entry_family": "research_only",
"observation": {"symbol": "TSLA"},
"signal_definition": {"conditions": ["reaction_1d=-0.132"]},
}
],
hints=[],
)
assert concept["strategy_design"]["export_ready_v1"] is False
assert concept["strategy_design"]["recommended_entry_family"] is None
def _setup_main_fixtures(tmp_path: Path) -> tuple[Path, Path]:
"""Create minimal ticket + hints fixture files for main() tests."""
tickets_dir = tmp_path / "tickets"
tickets_dir.mkdir()
ticket = {
"id": "edge_auto_test_1",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"priority_score": 70.0,
"entry_family": "pivot_breakout",
"observation": {"symbol": "XP"},
"date": "2026-02-20",
}
(tickets_dir / "ticket_1.yaml").write_text(yaml.safe_dump(ticket), encoding="utf-8")
hints_path = tmp_path / "hints.yaml"
hints_payload = {
"hints": [
{
"title": "Seasonal buyback blackout",
"observation": "Calendar-driven supply gap",
"hypothesis_type": "calendar_anomaly",
"symbols": ["SPY"],
"regime_bias": "Neutral",
"mechanism_tag": "flow",
}
]
}
hints_path.write_text(yaml.safe_dump(hints_payload), encoding="utf-8")
return tickets_dir, hints_path
def test_main_promote_hints_source_metadata(tmp_path: Path, monkeypatch) -> None:
"""Test that --promote-hints populates source metadata correctly."""
tickets_dir, hints_path = _setup_main_fixtures(tmp_path)
output_path = tmp_path / "concepts.yaml"
# Run WITH --promote-hints
monkeypatch.setattr(
"sys.argv",
[
"synthesize_edge_concepts.py",
"--tickets-dir",
str(tickets_dir),
"--hints",
str(hints_path),
"--output",
str(output_path),
"--promote-hints",
],
)
assert sec.main() == 0
result = yaml.safe_load(output_path.read_text())
assert result["source"]["promote_hints"] is True
assert result["source"]["real_ticket_count"] == 1
assert result["source"]["synthetic_ticket_count"] == 1
# Run WITHOUT --promote-hints
output_path2 = tmp_path / "concepts_no_promote.yaml"
monkeypatch.setattr(
"sys.argv",
[
"synthesize_edge_concepts.py",
"--tickets-dir",
str(tickets_dir),
"--hints",
str(hints_path),
"--output",
str(output_path2),
],
)
assert sec.main() == 0
result2 = yaml.safe_load(output_path2.read_text())
assert "promote_hints" not in result2["source"]
assert "real_ticket_count" not in result2["source"]
assert "synthetic_ticket_count" not in result2["source"]
def test_cap_synthetic_tickets_limits_count() -> None:
"""cap_synthetic_tickets should limit synthetic count to max(real_count * ratio, floor)."""
real_tickets = [
{
"id": f"edge_auto_real_{i}",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"priority_score": 70.0 + i,
"entry_family": "pivot_breakout",
}
for i in range(3)
]
synthetic_tickets = [
{
"id": f"hint_promo_test_{i}",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"priority_score": 30.0,
"entry_family": "research_only",
"_synthetic": True,
}
for i in range(10)
]
# ratio=1.5 → max_synthetic = max(3*1.5, 3) = 4 (ceil of 4.5)
capped = sec.cap_synthetic_tickets(
real_tickets=real_tickets,
synthetic_tickets=synthetic_tickets,
max_ratio=1.5,
floor=3,
)
assert len(capped) == 5 # ceil(3 * 1.5) = 5
def test_cap_synthetic_tickets_floor_applies() -> None:
"""When real_count * ratio < floor, use floor."""
real_tickets = [
{
"id": "edge_auto_real_0",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"priority_score": 70.0,
"entry_family": "pivot_breakout",
}
]
synthetic_tickets = [
{
"id": f"hint_promo_test_{i}",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"priority_score": 30.0,
"_synthetic": True,
}
for i in range(10)
]
# ratio=1.5 → max_synthetic = max(ceil(1*1.5), 3) = max(2, 3) = 3
capped = sec.cap_synthetic_tickets(
real_tickets=real_tickets,
synthetic_tickets=synthetic_tickets,
max_ratio=1.5,
floor=3,
)
assert len(capped) == 3
def test_cap_synthetic_tickets_no_truncation_needed() -> None:
"""When synthetic count is already within limit, no truncation."""
real_tickets = [{"id": f"r{i}", "priority_score": 70.0} for i in range(5)]
synthetic_tickets = [
{"id": f"s{i}", "priority_score": 30.0, "_synthetic": True} for i in range(2)
]
capped = sec.cap_synthetic_tickets(
real_tickets=real_tickets,
synthetic_tickets=synthetic_tickets,
max_ratio=1.5,
floor=3,
)
assert len(capped) == 2 # 2 < max(ceil(5*1.5), 3) = max(8, 3) = 8
def test_main_max_synthetic_ratio(tmp_path: Path, monkeypatch) -> None:
"""Test that --max-synthetic-ratio caps synthetic tickets in main()."""
tickets_dir = tmp_path / "tickets"
tickets_dir.mkdir()
# Create 2 real tickets
for i in range(2):
ticket = {
"id": f"edge_auto_test_{i}",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"priority_score": 70.0 + i,
"entry_family": "pivot_breakout",
"observation": {"symbol": f"SYM{i}"},
"date": "2026-02-20",
}
(tickets_dir / f"ticket_{i}.yaml").write_text(yaml.safe_dump(ticket), encoding="utf-8")
# Create hints with 8 entries
hints_path = tmp_path / "hints.yaml"
hints_payload = {
"hints": [
{
"title": f"Hint {i}",
"observation": f"Signal {i}",
"hypothesis_type": "breakout",
"symbols": [f"H{i}"],
"regime_bias": "RiskOn",
"mechanism_tag": "behavior",
}
for i in range(8)
]
}
hints_path.write_text(yaml.safe_dump(hints_payload), encoding="utf-8")
output_path = tmp_path / "concepts.yaml"
monkeypatch.setattr(
"sys.argv",
[
"synthesize_edge_concepts.py",
"--tickets-dir",
str(tickets_dir),
"--hints",
str(hints_path),
"--output",
str(output_path),
"--promote-hints",
"--max-synthetic-ratio",
"1.5",
],
)
assert sec.main() == 0
result = yaml.safe_load(output_path.read_text())
# 2 real tickets, ratio=1.5, floor=3 → max_synthetic = max(ceil(2*1.5), 3) = 3
assert result["source"]["synthetic_ticket_count"] == 3
def test_main_promote_hints_zero_promotions(tmp_path: Path, monkeypatch) -> None:
"""Test that --promote-hints with no hints still outputs promote_hints=True."""
tickets_dir = tmp_path / "tickets"
tickets_dir.mkdir()
ticket = {
"id": "edge_auto_test_1",
"hypothesis_type": "breakout",
"mechanism_tag": "behavior",
"regime": "RiskOn",
"priority_score": 70.0,
"entry_family": "pivot_breakout",
"observation": {"symbol": "XP"},
"date": "2026-02-20",
}
(tickets_dir / "ticket_1.yaml").write_text(yaml.safe_dump(ticket), encoding="utf-8")
output_path = tmp_path / "concepts.yaml"
# --promote-hints ON but no hints file provided
monkeypatch.setattr(
"sys.argv",
[
"synthesize_edge_concepts.py",
"--tickets-dir",
str(tickets_dir),
"--output",
str(output_path),
"--promote-hints",
],
)
assert sec.main() == 0
result = yaml.safe_load(output_path.read_text())
assert result["source"]["promote_hints"] is True
assert result["source"]["real_ticket_count"] == 1
assert result["source"]["synthetic_ticket_count"] == 0
# ---------------------------------------------------------------------------
# Helpers for merge / dedup tests
# ---------------------------------------------------------------------------
def _make_merge_concept(
concept_id: str,
hypothesis_type: str,
mechanism_tag: str,
ticket_count: int = 2,
avg_priority: float = 70.0,
symbols: list[str] | None = None,
conditions: list[str] | None = None,
entry_family: str | None = "pivot_breakout",
real_ticket_count: int | None = None,
synthetic_ticket_count: int | None = None,
synthetic_ticket_ids: list[str] | None = None,
) -> dict:
"""Create a concept dict for merge/dedup tests."""
if symbols is None:
symbols = ["XP"]
if conditions is None:
conditions = ["close > high20_prev", "rel_volume >= 1.5"]
support: dict = {
"ticket_count": ticket_count,
"avg_priority_score": avg_priority,
"symbols": symbols,
"entry_family_distribution": {entry_family: ticket_count} if entry_family else {},
"representative_conditions": conditions,
}
if real_ticket_count is not None:
support["real_ticket_count"] = real_ticket_count
if synthetic_ticket_count is not None:
support["synthetic_ticket_count"] = synthetic_ticket_count
evidence: dict = {
"ticket_ids": [f"ticket_{i}" for i in range(ticket_count)],
"matched_hint_titles": [],
}
if synthetic_ticket_ids:
evidence["synthetic_ticket_ids"] = synthetic_ticket_ids
concept = {
"id": concept_id,
"title": f"Test {hypothesis_type}",
"hypothesis_type": hypothesis_type,
"mechanism_tag": mechanism_tag,
"regime": "RiskOn",
"support": support,
"abstraction": {
"thesis": f"Test thesis for {hypothesis_type}",
"invalidation_signals": ["Signal 1", "Signal 2"],
},
"strategy_design": {
"playbooks": ["test_playbook"],
"recommended_entry_family": entry_family,
"export_ready_v1": entry_family in sec.DEFAULT_EXPORTABLE_FAMILIES
if entry_family
else False,
},
"evidence": evidence,
}
return concept
# ---------------------------------------------------------------------------
# condition_overlap_ratio tests
# ---------------------------------------------------------------------------
def test_overlap_ratio_identical():
"""Identical conditions should return 1.0."""
assert sec.condition_overlap_ratio(["A > 1", "B < 2"], ["A > 1", "B < 2"]) == 1.0
def test_overlap_ratio_disjoint():
"""Completely different conditions should return 0.0."""
assert sec.condition_overlap_ratio(["A > 1"], ["B < 2"]) == 0.0
def test_overlap_ratio_partial():
"""Partial overlap should return correct ratio."""
# |{a,b} ∩ {b,c}| / min(2,2) = 1/2 = 0.5
assert sec.condition_overlap_ratio(["A > 1", "B < 2"], ["B < 2", "C == 3"]) == 0.5
def test_overlap_ratio_empty():
"""Both empty should return 0.0."""
assert sec.condition_overlap_ratio([], []) == 0.0
def test_overlap_ratio_one_empty():
"""One empty should return 0.0."""
assert sec.condition_overlap_ratio(["A > 1"], []) == 0.0
assert sec.condition_overlap_ratio([], ["B < 2"]) == 0.0
def test_overlap_ratio_case_insensitive():
"""Comparison should be case-insensitive."""
assert sec.condition_overlap_ratio(["Close > MA50"], ["close > ma50"]) == 1.0
def test_overlap_ratio_asymmetric_containment():
"""Containment uses min(|A|,|B|): small set fully in large set -> 1.0."""
# A={a,b,c,d}, B={a,b,c} -> |A∩B|/min(4,3) = 3/3 = 1.0
assert (
sec.condition_overlap_ratio(
["A > 1", "B < 2", "C == 3", "D != 4"],
["A > 1", "B < 2", "C == 3"],
)
== 1.0
)
# ---------------------------------------------------------------------------
# merge_concepts tests
# ---------------------------------------------------------------------------
def test_merge_concepts_basic():
"""Merge two concepts with different mechanism_tag."""
primary = _make_merge_concept(
"c1",
"breakout",
"behavior",
ticket_count=3,
avg_priority=70.0,
symbols=["XP", "NOK"],
conditions=["close > high20_prev", "rel_volume >= 1.5"],
)
secondary = _make_merge_concept(
"c2",
"breakout",
"flow",
ticket_count=2,
avg_priority=60.0,
symbols=["NOK", "AAPL"],
conditions=["close > high20_prev", "volume > 2x"],
)
merged = sec.merge_concepts(primary, secondary)
assert merged["mechanism_tag"] == "behavior+flow"
assert merged["support"]["ticket_count"] == 5
assert "XP" in merged["support"]["symbols"]
assert "AAPL" in merged["support"]["symbols"]
assert "c2" in merged.get("merged_from", [])
def test_merge_concepts_entry_family_adoption():
"""If primary has no entry_family, adopt secondary's."""
primary = _make_merge_concept(
"c1",
"breakout",
"behavior",
ticket_count=3,
entry_family=None,
)
secondary = _make_merge_concept(
"c2",
"breakout",
"flow",
ticket_count=2,
entry_family="pivot_breakout",
)
merged = sec.merge_concepts(primary, secondary)
assert merged["strategy_design"]["recommended_entry_family"] == "pivot_breakout"
def test_merge_concepts_keeps_primary():
"""Primary's title, thesis, invalidation_signals should be preserved."""
primary = _make_merge_concept("c1", "breakout", "behavior", ticket_count=3)
secondary = _make_merge_concept("c2", "breakout", "flow", ticket_count=2)
merged = sec.merge_concepts(primary, secondary)
assert merged["abstraction"]["thesis"] == primary["abstraction"]["thesis"]
def test_merge_concepts_synthetic_fields():
"""If both have synthetic ticket counts, they should be summed."""
primary = _make_merge_concept(
"c1",
"breakout",
"behavior",
ticket_count=3,
real_ticket_count=2,
synthetic_ticket_count=1,
synthetic_ticket_ids=["hint_promo_1"],
)
secondary = _make_merge_concept(
"c2",
"breakout",
"flow",
ticket_count=2,
real_ticket_count=1,
synthetic_ticket_count=1,
synthetic_ticket_ids=["hint_promo_2"],
)
merged = sec.merge_concepts(primary, secondary)
assert merged["support"].get("real_ticket_count") == 3
assert merged["support"].get("synthetic_ticket_count") == 2
assert "hint_promo_1" in merged["evidence"].get("synthetic_ticket_ids", [])
assert "hint_promo_2" in merged["evidence"].get("synthetic_ticket_ids", [])
def test_merge_concepts_id_regeneration():
"""Merged concept should have regenerated id."""
primary = _make_merge_concept("old_id_1", "breakout", "behavior", ticket_count=3)
secondary = _make_merge_concept("old_id_2", "breakout", "flow", ticket_count=2)
merged = sec.merge_concepts(primary, secondary)
# mechanism_tag becomes "behavior+flow"
expected_id = sec.sanitize_identifier("edge_concept_breakout_behavior_flow_RiskOn")
assert merged["id"] == expected_id
# ---------------------------------------------------------------------------
# deduplicate_concepts tests
# ---------------------------------------------------------------------------
def test_dedup_high_overlap_merge():
"""Two concepts with high overlap should be merged."""
c1 = _make_merge_concept(
"c1",
"breakout",
"behavior",
ticket_count=3,
conditions=["close > high20", "volume > 2x", "RSI > 50"],
)
c2 = _make_merge_concept(
"c2",
"breakout",
"flow",
ticket_count=2,
conditions=["close > high20", "volume > 2x"],
)
result, count = sec.deduplicate_concepts([c1, c2], overlap_threshold=0.75)
assert len(result) == 1
assert count == 1
def test_dedup_keeps_distinct():
"""Two concepts with low overlap should NOT be merged."""
c1 = _make_merge_concept(
"c1",
"breakout",
"behavior",
ticket_count=3,
conditions=["close > high20", "volume > 2x"],
)
c2 = _make_merge_concept(
"c2",
"breakout",
"flow",
ticket_count=2,
conditions=["RSI < 30", "price < SMA200"],
)
result, count = sec.deduplicate_concepts([c1, c2], overlap_threshold=0.75)
assert len(result) == 2
assert count == 0
def test_dedup_different_hypothesis_never_merge():
"""Concepts with different hypothesis_type should never be merged."""
c1 = _make_merge_concept(
"c1",
"breakout",
"behavior",
ticket_count=3,
conditions=["close > high20", "volume > 2x"],
)
c2 = _make_merge_concept(
"c2",
"panic_reversal",
"behavior",
ticket_count=2,
conditions=["close > high20", "volume > 2x"],
)
result, count = sec.deduplicate_concepts([c1, c2], overlap_threshold=0.5)
assert len(result) == 2
assert count == 0
def test_dedup_single_concept():
"""Single concept should pass through unchanged."""
c1 = _make_merge_concept("c1", "breakout", "behavior", ticket_count=3)
result, count = sec.deduplicate_concepts([c1], overlap_threshold=0.75)
assert len(result) == 1
assert count == 0
def test_dedup_empty_list():
"""Empty list should return empty."""
result, count = sec.deduplicate_concepts([], overlap_threshold=0.75)
assert len(result) == 0
assert count == 0
# ---------------------------------------------------------------------------
# main() integration tests for dedup
# ---------------------------------------------------------------------------
def test_main_no_dedup_flag(tmp_path: Path, monkeypatch) -> None:
"""--no-dedup should disable deduplication."""
tickets_dir, hints_path = _setup_main_fixtures(tmp_path)
# Add a second ticket to create potential duplicate
ticket2 = {
"id": "edge_auto_test_2",
"hypothesis_type": "breakout",
"mechanism_tag": "flow",
"regime": "RiskOn",
"priority_score": 65.0,
"entry_family": "pivot_breakout",
"observation": {"symbol": "XP"},
"signal_definition": {"conditions": ["close > high20_prev", "rel_volume >= 1.5"]},
"date": "2026-02-20",
}
(tickets_dir / "ticket_2.yaml").write_text(yaml.safe_dump(ticket2), encoding="utf-8")
output_path = tmp_path / "concepts.yaml"
monkeypatch.setattr(
"sys.argv",
[
"synthesize_edge_concepts.py",
"--tickets-dir",
str(tickets_dir),
"--output",
str(output_path),
"--no-dedup",
],
)
assert sec.main() == 0
result = yaml.safe_load(output_path.read_text())
assert result["source"]["dedup_enabled"] is False
assert result["source"]["dedup_merged_count"] == 0
def test_main_overlap_threshold_effect(tmp_path: Path, monkeypatch) -> None:
"""--overlap-threshold should affect merge behavior."""
tickets_dir, _ = _setup_main_fixtures(tmp_path)
# Create two tickets in different mechanism groups with same conditions
for i, mech in enumerate(["behavior", "flow"]):
ticket = {
"id": f"edge_auto_dup_{i}",
"hypothesis_type": "breakout",
"mechanism_tag": mech,
"regime": "RiskOn",
"priority_score": 70.0,
"entry_family": "pivot_breakout",
"observation": {"symbol": "XP"},
"signal_definition": {"conditions": ["close > high20_prev", "rel_volume >= 1.5"]},
"date": "2026-02-20",
}
(tickets_dir / f"ticket_dup_{i}.yaml").write_text(yaml.safe_dump(ticket), encoding="utf-8")
output_path = tmp_path / "concepts.yaml"
# With low threshold -> should merge
monkeypatch.setattr(
"sys.argv",
[
"synthesize_edge_concepts.py",
"--tickets-dir",
str(tickets_dir),
"--output",
str(output_path),
"--overlap-threshold",
"0.5",
],
)
assert sec.main() == 0
result = yaml.safe_load(output_path.read_text())
assert result["source"]["dedup_enabled"] is True
assert result["source"]["overlap_threshold"] == 0.5