Hypotheses API

asrquant.hypotheses

Hypothesis discovery, search and audit for ASRQuant 1.2.0.

This module turns data, literature, model disagreement and robustness evidence into research candidates. It is intentionally conservative: statistical screening may suggest a useful hypothesis, but it never establishes scientific novelty or causality automatically.

The public entry points are::

asr.hypotheses.from_data(...)
asr.hypotheses.from_literature(...)
asr.hypotheses.from_model_disagreement(...)
asr.hypotheses.from_robustness(...)
asr.hypotheses.discover(...)
asr.hypotheses.search(...)
asr.hypotheses.audit(...)

Every data-driven screen records the number of tests performed, discovery and holdout samples, raw p-values and Benjamini-Hochberg q-values when applicable.

HypothesisIdea dataclass

One falsifiable research hypothesis with separate evidence and novelty states.

Source code in src/asrquant/hypotheses.py

@dataclass
class HypothesisIdea:
    """One falsifiable research hypothesis with separate evidence and novelty states."""

    hypothesis_id: str
    statement: str
    research_question: str
    domain: str = "quantitative_finance"
    source: str = "data"
    data_status: str = "EXPLORATORY"
    novelty_status: str = "NOVELTY_NOT_ESTABLISHED"
    evidence_status: str = "PROPOSED"
    priority_score: float = 0.5
    predictor: str | None = None
    target: str | None = None
    expected_sign: str | None = None
    horizon: int | str | None = None
    mechanism: str = ""
    null_hypothesis: str = "No stable out-of-sample relationship under the pre-specified test."
    falsification_rule: str = "Downgrade or reject the hypothesis if the effect fails chronology-safe holdout and robustness checks."
    methods: tuple[str, ...] = ()
    data_requirements: tuple[str, ...] = ()
    alternative_explanations: tuple[str, ...] = ()
    source_observations: tuple[str, ...] = ()
    evidence: dict[str, Any] = field(default_factory=dict)
    references: list[dict[str, Any]] = field(default_factory=list)
    metadata: dict[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        if self.data_status not in DATA_STATUSES:
            raise InputValidationError(f"unknown data_status {self.data_status!r}")
        if self.novelty_status not in NOVELTY_STATUSES:
            raise InputValidationError(f"unknown novelty_status {self.novelty_status!r}")
        self.priority_score = float(np.clip(self.priority_score, 0.0, 1.0))

    @property
    def summary(self) -> pd.Series:
        return pd.Series(
            {
                "hypothesis_id": self.hypothesis_id,
                "domain": self.domain,
                "source": self.source,
                "data_status": self.data_status,
                "novelty_status": self.novelty_status,
                "evidence_status": self.evidence_status,
                "priority_score": self.priority_score,
                "predictor": self.predictor,
                "target": self.target,
                "expected_sign": self.expected_sign,
                "horizon": self.horizon,
            }
        )

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)

    def to_frame(self) -> pd.DataFrame:
        return self.summary.rename("value").to_frame()

    def start(self, *, name: str | None = None) -> ResearchProject:
        """Hand this hypothesis to ASRQuant's existing end-to-end ResearchProject."""
        return ResearchProject.from_hypothesis(
            self.statement,
            name=name or f"ASRQuant Research — {self.hypothesis_id}",
            topic=self.domain,
            predictor=self.predictor,
            target=self.target,
            expected_sign=self.expected_sign,
            horizon=self.horizon,
            novelty_status=self.novelty_status.lower(),
            evidence_status=self.data_status.lower(),
            mechanism=self.mechanism,
            invalidation_criteria=[self.falsification_rule],
            metadata={
                "hypothesis_source": self.source,
                "hypothesis_priority": self.priority_score,
                "research_question": self.research_question,
                "methods": list(self.methods),
                "data_requirements": list(self.data_requirements),
                "alternative_explanations": list(self.alternative_explanations),
                "source_observations": list(self.source_observations),
                "evidence": self.evidence,
                "references": self.references,
                **self.metadata,
            },
        )

start

start(*, name: str | None = None) -> ResearchProject

Hand this hypothesis to ASRQuant's existing end-to-end ResearchProject.

Source code in src/asrquant/hypotheses.py

def start(self, *, name: str | None = None) -> ResearchProject:
    """Hand this hypothesis to ASRQuant's existing end-to-end ResearchProject."""
    return ResearchProject.from_hypothesis(
        self.statement,
        name=name or f"ASRQuant Research — {self.hypothesis_id}",
        topic=self.domain,
        predictor=self.predictor,
        target=self.target,
        expected_sign=self.expected_sign,
        horizon=self.horizon,
        novelty_status=self.novelty_status.lower(),
        evidence_status=self.data_status.lower(),
        mechanism=self.mechanism,
        invalidation_criteria=[self.falsification_rule],
        metadata={
            "hypothesis_source": self.source,
            "hypothesis_priority": self.priority_score,
            "research_question": self.research_question,
            "methods": list(self.methods),
            "data_requirements": list(self.data_requirements),
            "alternative_explanations": list(self.alternative_explanations),
            "source_observations": list(self.source_observations),
            "evidence": self.evidence,
            "references": self.references,
            **self.metadata,
        },
    )

HypothesisAuditResult dataclass

Prior-art/evidence audit that never auto-asserts global novelty.

Source code in src/asrquant/hypotheses.py

@dataclass
class HypothesisAuditResult:
    """Prior-art/evidence audit that never auto-asserts global novelty."""

    hypothesis_id: str
    novelty_status: str
    data_status: str
    closest_matches: pd.DataFrame
    recommendation: str
    warnings: tuple[str, ...] = ()
    corpus_fingerprint: str | None = None

    @property
    def summary(self) -> pd.Series:
        best_similarity = (
            float(self.closest_matches["similarity"].max())
            if len(self.closest_matches) and "similarity" in self.closest_matches
            else np.nan
        )
        return pd.Series(
            {
                "hypothesis_id": self.hypothesis_id,
                "novelty_status": self.novelty_status,
                "data_status": self.data_status,
                "matches": int(len(self.closest_matches)),
                "best_similarity": best_similarity,
                "recommendation": self.recommendation,
            }
        )

    def to_frame(self) -> pd.DataFrame:
        return self.closest_matches.copy()

    def to_dict(self) -> dict[str, Any]:
        return {
            "result_type": "hypothesis_audit",
            "summary": self.summary.to_dict(),
            "matches": self.closest_matches.to_dict(orient="records"),
            "warnings": list(self.warnings),
            "corpus_fingerprint": self.corpus_fingerprint,
        }

HypothesisSearchResult dataclass

Search matches across generated hypotheses and/or supplied literature.

Source code in src/asrquant/hypotheses.py

@dataclass
class HypothesisSearchResult:
    """Search matches across generated hypotheses and/or supplied literature."""

    query: str
    hypotheses: pd.DataFrame = field(default_factory=pd.DataFrame)
    excerpts: pd.DataFrame = field(default_factory=pd.DataFrame)

    @property
    def summary(self) -> pd.Series:
        return pd.Series(
            {
                "query": self.query,
                "hypothesis_matches": int(len(self.hypotheses)),
                "source_excerpts": int(len(self.excerpts)),
            }
        )

    def to_frame(self) -> pd.DataFrame:
        return self.hypotheses.copy()

    def to_dict(self) -> dict[str, Any]:
        return {
            "result_type": "hypothesis_search",
            "summary": self.summary.to_dict(),
            "hypotheses": self.hypotheses.to_dict(orient="records"),
            "excerpts": self.excerpts.to_dict(orient="records"),
        }

HypothesisCollection dataclass

Ranked, searchable hypothesis set with screening provenance.

Source code in src/asrquant/hypotheses.py

@dataclass
class HypothesisCollection:
    """Ranked, searchable hypothesis set with screening provenance."""

    hypotheses: list[HypothesisIdea]
    metadata: dict[str, Any] = field(default_factory=dict)

    def __len__(self) -> int:
        return len(self.hypotheses)

    def __iter__(self):
        return iter(self.hypotheses)

    def __getitem__(self, item: int) -> HypothesisIdea:
        return self.hypotheses[item]

    @property
    def summary(self) -> pd.Series:
        counts = pd.Series([h.data_status for h in self.hypotheses]).value_counts()
        return pd.Series(
            {
                "hypotheses": len(self.hypotheses),
                "tests_performed": int(self.metadata.get("tests_performed", 0)),
                "multiple_testing": self.metadata.get("multiple_testing", "not_applicable"),
                "out_of_sample_supported": int(counts.get("OUT_OF_SAMPLE_SUPPORTED", 0)),
                "data_supported": int(counts.get("DATA_SUPPORTED", 0)),
                "exploratory": int(counts.get("EXPLORATORY", 0)),
                "falsified": int(counts.get("FALSIFIED", 0)),
            }
        )

    def select(self, identifier: str | int) -> HypothesisIdea:
        if isinstance(identifier, int):
            return self.hypotheses[identifier]
        for item in self.hypotheses:
            if item.hypothesis_id == identifier:
                return item
        raise KeyError(f"unknown hypothesis {identifier!r}")

    def to_frame(self) -> pd.DataFrame:
        rows = []
        for item in self.hypotheses:
            row = {
                "hypothesis_id": item.hypothesis_id,
                "statement": item.statement,
                "research_question": item.research_question,
                "domain": item.domain,
                "source": item.source,
                "data_status": item.data_status,
                "novelty_status": item.novelty_status,
                "evidence_status": item.evidence_status,
                "priority_score": item.priority_score,
                "predictor": item.predictor,
                "target": item.target,
                "expected_sign": item.expected_sign,
                "horizon": item.horizon,
                "q_value": item.evidence.get("q_value", np.nan),
                "holdout_p_value": item.evidence.get("holdout_p_value", np.nan),
                "discovery_effect": item.evidence.get("discovery_effect", np.nan),
                "holdout_effect": item.evidence.get("holdout_effect", np.nan),
            }
            rows.append(row)
        return pd.DataFrame(rows)

    def to_dict(self) -> dict[str, Any]:
        return {
            "result_type": "hypothesis_collection",
            "summary": self.summary.to_dict(),
            "metadata": dict(self.metadata),
            "hypotheses": [item.to_dict() for item in self.hypotheses],
        }

    def rank(self, *, by: str = "priority_score", ascending: bool = False) -> "HypothesisCollection":
        key = str(by)
        if key == "priority_score":
            ranked = sorted(self.hypotheses, key=lambda item: item.priority_score, reverse=not ascending)
        elif key in {"q_value", "holdout_p_value"}:
            ranked = sorted(
                self.hypotheses,
                key=lambda item: float(item.evidence.get(key, np.inf)),
                reverse=ascending,
            )
        else:
            raise InputValidationError("by must be priority_score, q_value, or holdout_p_value")
        return HypothesisCollection(ranked, dict(self.metadata))

    def search(self, query: str, *, top_k: int = 10, min_similarity: float = 0.0) -> pd.DataFrame:
        if not str(query).strip():
            raise InputValidationError("query must not be empty")
        rows = []
        for item in self.hypotheses:
            score = _text_similarity(query, f"{item.research_question} {item.statement}")
            if score >= min_similarity:
                rows.append(
                    {
                        "hypothesis_id": item.hypothesis_id,
                        "similarity": score,
                        "statement": item.statement,
                        "data_status": item.data_status,
                        "novelty_status": item.novelty_status,
                        "priority_score": item.priority_score,
                    }
                )
        if not rows:
            return pd.DataFrame(
                columns=["hypothesis_id", "similarity", "statement", "data_status", "novelty_status", "priority_score"]
            )
        return pd.DataFrame(rows).sort_values(["similarity", "priority_score"], ascending=False).head(top_k).reset_index(drop=True)

    def audit(
        self,
        identifier: str | int,
        *,
        corpus: LiteratureCorpus | HypothesisRegistry | str | Path | Sequence[Any] | None = None,
        topic: str | None = None,
        top_k: int = 10,
    ) -> HypothesisAuditResult:
        return audit(self.select(identifier), corpus=corpus, topic=topic, top_k=top_k)

    def start(self, identifier: str | int, *, name: str | None = None) -> ResearchProject:
        return self.select(identifier).start(name=name)

from_data

from_data(data: DataFrame | Series | Mapping[str, Any], *, domain: str = 'quantitative_finance', targets: str | Sequence[str] | None = None, horizons: Sequence[int] = (1, 5, 20), lags: Sequence[int] = (0, 1, 5), transforms: Mapping[str, str] | None = None, holdout_fraction: float = 0.3, min_observations: int = 80, fdr_alpha: float = 0.05, min_abs_effect: float = 0.08, max_tests: int = 1500, max_candidates: int = 50, include_regime_tests: bool = True, include_cointegration: bool = True, include_structural_scan: bool = True) -> HypothesisCollection

Discover falsifiable hypotheses directly from time-indexed quantitative data.

The function uses a chronological discovery/holdout split and Benjamini-Hochberg correction across the statistical screening family. Returned candidates are still research hypotheses, not established findings.

Source code in src/asrquant/hypotheses.py

def from_data(
    data: pd.DataFrame | pd.Series | Mapping[str, Any],
    *,
    domain: str = "quantitative_finance",
    targets: str | Sequence[str] | None = None,
    horizons: Sequence[int] = (1, 5, 20),
    lags: Sequence[int] = (0, 1, 5),
    transforms: Mapping[str, str] | None = None,
    holdout_fraction: float = 0.30,
    min_observations: int = 80,
    fdr_alpha: float = 0.05,
    min_abs_effect: float = 0.08,
    max_tests: int = 1_500,
    max_candidates: int = 50,
    include_regime_tests: bool = True,
    include_cointegration: bool = True,
    include_structural_scan: bool = True,
) -> HypothesisCollection:
    """Discover falsifiable hypotheses directly from time-indexed quantitative data.

    The function uses a chronological discovery/holdout split and Benjamini-Hochberg
    correction across the statistical screening family.  Returned candidates are
    still *research hypotheses*, not established findings.
    """
    if not 0.05 <= holdout_fraction <= 0.5:
        raise InputValidationError("holdout_fraction must be between 0.05 and 0.5")
    if min_observations < 20:
        raise InputValidationError("min_observations must be at least 20")
    if not 0 < fdr_alpha < 1:
        raise InputValidationError("fdr_alpha must lie in (0, 1)")
    if min_abs_effect < 0:
        raise InputValidationError("min_abs_effect must be non-negative")
    if max_tests < 1 or max_candidates < 1:
        raise InputValidationError("max_tests and max_candidates must be positive")

    frame, source_map = _coerce_panel(data)
    active_domain = _normalise_domain(domain)
    if len(frame) < min_observations:
        raise InputValidationError(f"data-driven discovery requires at least {min_observations} observations")

    transform_map: dict[str, str] = {}
    transformed = pd.DataFrame(index=frame.index)
    supplied = {str(key): str(value) for key, value in (transforms or {}).items()}
    for column in frame.columns:
        method = supplied.get(str(column), _infer_transform(frame[column], active_domain))
        transform_map[str(column)] = method
        transformed[str(column)] = _transform_series(frame[column], method)

    if targets is None:
        target_names = [str(column) for column in frame.columns]
    elif isinstance(targets, str):
        target_names = [targets]
    else:
        target_names = [str(item) for item in targets]
    unknown = sorted(set(target_names) - set(map(str, frame.columns)))
    if unknown:
        raise InputValidationError(f"targets not found in data: {unknown}")

    horizon_values = tuple(sorted({int(value) for value in horizons}))
    lag_values = tuple(sorted({int(value) for value in lags}))
    if not horizon_values or any(value <= 0 for value in horizon_values):
        raise InputValidationError("horizons must contain positive integers")
    if not lag_values or any(value < 0 for value in lag_values):
        raise InputValidationError("lags must contain non-negative integers")

    tests: list[dict[str, Any]] = []
    tests.extend(
        _correlation_tests(
            transformed,
            transform_map,
            targets=target_names,
            horizons=horizon_values,
            lags=lag_values,
            holdout_fraction=holdout_fraction,
            min_observations=min_observations,
            max_tests=max_tests,
        )
    )
    remaining = max(0, max_tests - len(tests))
    if include_regime_tests and remaining:
        tests.extend(
            _regime_tests(
                transformed,
                transform_map,
                targets=target_names,
                horizons=horizon_values,
                holdout_fraction=holdout_fraction,
                min_observations=min_observations,
                max_tests=remaining,
            )
        )
    remaining = max(0, max_tests - len(tests))
    if include_cointegration and remaining and frame.shape[1] >= 2:
        tests.extend(
            _cointegration_tests(
                frame,
                holdout_fraction=holdout_fraction,
                min_observations=min_observations,
                max_tests=remaining,
            )
        )

    ideas: list[HypothesisIdea] = []
    if tests:
        q_values = _bh_adjust([float(row["p_value"]) for row in tests])
        for row, q_value in zip(tests, q_values):
            effect = float(row["discovery_effect"])
            # Keep FDR-supported tests even when the effect is small; otherwise apply a practical floor.
            if abs(effect) < min_abs_effect and q_value > fdr_alpha:
                continue
            ideas.append(
                _idea_from_test(
                    row,
                    q_value=float(q_value),
                    alpha=fdr_alpha,
                    domain=active_domain,
                    transforms=transform_map,
                    tests_performed=len(tests),
                )
            )

    structural_ideas: list[HypothesisIdea] = []
    if include_structural_scan:
        try:
            board = _discovery.weekly(
                data=frame,
                domain=active_domain,
                n=max_candidates,
                include_catalog=False,
            )
            for candidate in board.candidates:
                structural_ideas.append(
                    _idea_from_research_candidate(
                        candidate, source="data_structural_scan", data_status="EXPLORATORY"
                    )
                )
        except (ValueError, TypeError, KeyError, np.linalg.LinAlgError):
            structural_ideas = []

        # Conservative fallback: even when threshold-based discovery finds no
        # event, keep one explicitly exploratory structural question so the
        # structural-scan channel remains visible and auditable. This does not
        # assert statistical support or novelty.
        if not structural_ideas and len(frame) >= min_observations:
            numeric = frame.select_dtypes(include=[np.number]).dropna(how="all")
            if numeric.shape[1] >= 2:
                corr = numeric.corr().abs()
                upper = corr.where(np.triu(np.ones(corr.shape), k=1).astype(bool)).stack()
                if len(upper):
                    left, right = upper.idxmax()
                    pair = numeric[[left, right]].dropna()
                    half = max(1, len(pair) // 2)
                    c1 = float(pair.iloc[:half].corr().iloc[0, 1]) if half >= 3 else np.nan
                    c2 = float(pair.iloc[half:].corr().iloc[0, 1]) if len(pair) - half >= 3 else np.nan
                    structural_ideas.append(HypothesisIdea(
                        hypothesis_id=_identifier("H-STRUCT", left, right, len(pair)),
                        statement=f"The dependence between {left} and {right} may vary across market states or sample regimes.",
                        research_question=f"Is the relationship between {left} and {right} structurally stable through time?",
                        domain=active_domain,
                        source="data_structural_scan",
                        data_status="EXPLORATORY",
                        novelty_status="NOVELTY_NOT_ESTABLISHED",
                        evidence_status="STRUCTURAL_SCREEN",
                        priority_score=float(min(0.60, 0.30 + 0.25 * float(upper.max()))),
                        predictor=str(left),
                        target=str(right),
                        mechanism="A structural-stability question was generated from the strongest observed pairwise dependence; no causal interpretation is implied.",
                        methods=("rolling correlation", "change-point tests", "subperiod stability", "block bootstrap"),
                        data_requirements=(str(left), str(right)),
                        evidence={"absolute_full_sample_correlation": float(upper.max()), "first_half_correlation": c1, "second_half_correlation": c2, "sample_size": int(len(pair))},
                        metadata={"test_type": "structural_stability_fallback", "threshold_triggered": False},
                    ))
            elif numeric.shape[1] == 1:
                column = str(numeric.columns[0])
                series = numeric.iloc[:, 0].dropna()
                structural_ideas.append(HypothesisIdea(
                    hypothesis_id=_identifier("H-STRUCT", column, len(series)),
                    statement=f"The distribution of {column} may be state-dependent rather than stable through time.",
                    research_question=f"Is the distribution of {column} structurally stable across the observed sample?",
                    domain=active_domain,
                    source="data_structural_scan",
                    data_status="EXPLORATORY",
                    novelty_status="NOVELTY_NOT_ESTABLISHED",
                    evidence_status="STRUCTURAL_SCREEN",
                    priority_score=0.30,
                    predictor=column,
                    methods=("rolling moments", "change-point tests", "subperiod stability", "bootstrap"),
                    data_requirements=(column,),
                    evidence={"sample_size": int(len(series))},
                    metadata={"test_type": "structural_stability_fallback", "threshold_triggered": False},
                ))
        ideas.extend(structural_ideas)

    ideas = _deduplicate(ideas)
    # Preserve evidence-channel provenance: if a structural candidate was merged
    # into a stronger statistical screen, retain the highest-priority structural
    # observation as an explicit exploratory candidate rather than making that
    # channel disappear from the result set.
    if structural_ideas and not any(item.source == "data_structural_scan" for item in ideas):
        retained = max(structural_ideas, key=lambda item: item.priority_score)
        retained.metadata["retained_for_source_provenance"] = True
        ideas.append(retained)
    ideas = sorted(ideas, key=lambda item: item.priority_score, reverse=True)[:max_candidates]
    return HypothesisCollection(
        ideas,
        metadata={
            "source": "data",
            "domain": active_domain,
            "rows": int(len(frame)),
            "columns": int(frame.shape[1]),
            "targets": target_names,
            "transforms": transform_map,
            "source_map": source_map,
            "tests_performed": int(len(tests)),
            "multiple_testing": "Benjamini-Hochberg FDR",
            "fdr_alpha": float(fdr_alpha),
            "holdout_fraction": float(holdout_fraction),
            "min_observations": int(min_observations),
            "max_tests": int(max_tests),
        },
    )

from_literature

from_literature(papers: LiteratureCorpus | HypothesisRegistry | str | Path | Sequence[Any], *, topic: str | None = None, max_candidates: int = 50) -> HypothesisCollection

Discover source-linked hypotheses and research gaps from scientific literature.

Source code in src/asrquant/hypotheses.py

def from_literature(
    papers: LiteratureCorpus | HypothesisRegistry | str | Path | Sequence[Any],
    *,
    topic: str | None = None,
    max_candidates: int = 50,
) -> HypothesisCollection:
    """Discover source-linked hypotheses and research gaps from scientific literature."""
    try:
        source = _as_literature_source(papers, topic=topic)
        board = _discovery.from_literature(source, topic=topic, max_candidates=max_candidates)
    except (ValueError, TypeError, OSError, RuntimeError) as exc:
        raise HypothesisDiscoveryError(f"literature hypothesis discovery failed: {exc}") from exc
    ideas = [
        _idea_from_research_candidate(candidate, source="literature", data_status="LITERATURE_DERIVED")
        for candidate in board.candidates
    ]
    # Translate corpus-relative labels to conservative public novelty states.
    for idea, candidate in zip(ideas, board.candidates):
        original = str(getattr(candidate, "novelty_status", "NOT_ESTABLISHED")).upper().replace("-", "_")
        if original == "CONTRADICTORY":
            idea.novelty_status = "CONTRADICTORY_LITERATURE"
        elif original == "CORPUS_NOVEL":
            idea.novelty_status = "POTENTIAL_GAP"
        elif original in {"ESTABLISHED", "REPLICATED"}:
            idea.novelty_status = "PRIOR_ART_FOUND"
        elif original == "UNDEREXPLORED":
            idea.novelty_status = "CORPUS_RELATED"
        else:
            idea.novelty_status = "NOVELTY_NOT_ESTABLISHED"
        idea.metadata["corpus_relative_label"] = original
    fingerprint = source.corpus_fingerprint if isinstance(source, HypothesisRegistry) else source.fingerprint
    return HypothesisCollection(
        sorted(ideas, key=lambda item: item.priority_score, reverse=True),
        metadata={
            "source": "literature",
            "topic": topic,
            "corpus_fingerprint": fingerprint,
            "tests_performed": 0,
            "multiple_testing": "not_applicable",
        },
    )

from_model_disagreement

from_model_disagreement(predictions: DataFrame | Mapping[str, Sequence[float]], *, domain: str = 'quantitative_finance', max_candidates: int = 25) -> HypothesisCollection

Generate hypotheses from periods where plausible models disagree materially.

Source code in src/asrquant/hypotheses.py

def from_model_disagreement(
    predictions: pd.DataFrame | Mapping[str, Sequence[float]],
    *,
    domain: str = "quantitative_finance",
    max_candidates: int = 25,
) -> HypothesisCollection:
    """Generate hypotheses from periods where plausible models disagree materially."""
    board = _discovery.weekly(
        predictions=pd.DataFrame(predictions),
        domain=domain,
        n=max_candidates,
        include_catalog=False,
    )
    ideas = [
        _idea_from_research_candidate(candidate, source="model_disagreement", data_status="EXPLORATORY")
        for candidate in board.candidates
    ]
    return HypothesisCollection(ideas, {"source": "model_disagreement", "tests_performed": 0})

from_robustness

from_robustness(results: DataFrame, *, metric: str, domain: str = 'quantitative_finance', max_candidates: int = 25) -> HypothesisCollection

Generate hypotheses from specification-sensitive research results.

Source code in src/asrquant/hypotheses.py

def from_robustness(
    results: pd.DataFrame,
    *,
    metric: str,
    domain: str = "quantitative_finance",
    max_candidates: int = 25,
) -> HypothesisCollection:
    """Generate hypotheses from specification-sensitive research results."""
    board = _discovery.weekly(
        robustness_results=pd.DataFrame(results),
        robustness_metric=metric,
        domain=domain,
        n=max_candidates,
        include_catalog=False,
    )
    ideas = [
        _idea_from_research_candidate(candidate, source="robustness", data_status="EXPLORATORY")
        for candidate in board.candidates
    ]
    return HypothesisCollection(ideas, {"source": "robustness", "metric": metric, "tests_performed": 0})

audit

audit(hypothesis: HypothesisIdea | str, *, corpus: LiteratureCorpus | HypothesisRegistry | str | Path | Sequence[Any] | None = None, topic: str | None = None, top_k: int = 10) -> HypothesisAuditResult

Audit prior art around a hypothesis without asserting global novelty.

Source code in src/asrquant/hypotheses.py

def audit(
    hypothesis: HypothesisIdea | str,
    *,
    corpus: LiteratureCorpus | HypothesisRegistry | str | Path | Sequence[Any] | None = None,
    topic: str | None = None,
    top_k: int = 10,
) -> HypothesisAuditResult:
    """Audit prior art around a hypothesis without asserting global novelty."""
    if isinstance(hypothesis, HypothesisIdea):
        idea = hypothesis
    else:
        statement = str(hypothesis).strip()
        if not statement:
            raise InputValidationError("hypothesis must not be empty")
        idea = HypothesisIdea(
            _identifier("H-AUDIT", statement),
            statement,
            f"Is the following hypothesis supported and distinct from documented prior work: {statement}?",
            domain=_normalise_domain(topic or "quantitative_finance"),
            source="manual",
            data_status="NOT_TESTED",
        )

    if corpus is None:
        return HypothesisAuditResult(
            idea.hypothesis_id,
            "NOVELTY_NOT_ESTABLISHED",
            idea.data_status,
            pd.DataFrame(columns=["hypothesis_id", "similarity", "statement", "corpus_label", "evidence_status", "source_count", "pages"]),
            "Supply a documented literature corpus and perform a manual prior-art review before making any novelty claim.",
            warnings=("No literature corpus was supplied; novelty cannot be assessed.",),
        )

    registry, fingerprint = _literature_registry(corpus, topic=topic or idea.domain)
    rows = []
    for item in registry.hypotheses:
        similarity = _text_similarity(idea.statement, item.statement)
        rows.append(
            {
                "hypothesis_id": item.hypothesis_id,
                "similarity": similarity,
                "statement": item.statement,
                "corpus_label": item.novelty_status,
                "evidence_status": item.evidence_status,
                "source_count": item.source_count,
                "pages": ", ".join(f"{excerpt.paper_id}:p{excerpt.page}" for excerpt in item.evidence),
            }
        )
    matches = (
        pd.DataFrame(rows).sort_values("similarity", ascending=False).head(top_k).reset_index(drop=True)
        if rows
        else pd.DataFrame(columns=["hypothesis_id", "similarity", "statement", "corpus_label", "evidence_status", "source_count", "pages"])
    )

    if matches.empty:
        novelty = "NOVELTY_NOT_ESTABLISHED"
        recommendation = "No hypothesis-like passage was extracted from this corpus. Expand the prior-art search; absence of a match is not evidence of novelty."
    else:
        best = matches.iloc[0]
        similarity = float(best.similarity)
        label = str(best.corpus_label).lower().replace("-", "_")
        if label == "contradictory" and similarity >= 0.45:
            novelty = "CONTRADICTORY_LITERATURE"
            recommendation = "Review the contradictory source passages and design a test that discriminates between competing explanations."
        elif similarity >= 0.72:
            novelty = "PRIOR_ART_FOUND"
            recommendation = "Close prior art was found. Frame the project as replication, extension, boundary test or methodological comparison unless broader review supports another claim."
        elif label == "corpus_novel" and similarity >= 0.45:
            novelty = "POTENTIAL_GAP"
            recommendation = "The supplied corpus contains a related explicit gap. Validate it against broader literature before describing it as novel."
        elif similarity >= 0.50:
            novelty = "CORPUS_RELATED"
            recommendation = "Related prior work exists. Compare definitions, data, horizon, method and market scope before formulating the contribution."
        else:
            novelty = "NOVELTY_NOT_ESTABLISHED"
            recommendation = "No close match was detected in this corpus. Expand the search across databases, synonyms and adjacent literatures before any novelty claim."

    warnings = (
        "Novelty is corpus-relative; this audit does not search the complete global literature.",
        "Automatically extracted source passages must be checked on the cited pages before publication.",
    )
    return HypothesisAuditResult(
        idea.hypothesis_id,
        novelty,
        idea.data_status,
        matches,
        recommendation,
        warnings=warnings,
        corpus_fingerprint=fingerprint,
    )

search

search(query: str, *, hypotheses: HypothesisCollection | Sequence[HypothesisIdea] | None = None, papers: LiteratureCorpus | HypothesisRegistry | str | Path | Sequence[Any] | None = None, topic: str | None = None, top_k: int = 10) -> HypothesisSearchResult

Search generated hypotheses and source-linked literature with one query.

Source code in src/asrquant/hypotheses.py

def search(
    query: str,
    *,
    hypotheses: HypothesisCollection | Sequence[HypothesisIdea] | None = None,
    papers: LiteratureCorpus | HypothesisRegistry | str | Path | Sequence[Any] | None = None,
    topic: str | None = None,
    top_k: int = 10,
) -> HypothesisSearchResult:
    """Search generated hypotheses and source-linked literature with one query."""
    if not str(query).strip():
        raise InputValidationError("query must not be empty")
    if hypotheses is None:
        hypothesis_matches = pd.DataFrame()
    else:
        collection = hypotheses if isinstance(hypotheses, HypothesisCollection) else HypothesisCollection(list(hypotheses))
        hypothesis_matches = collection.search(query, top_k=top_k)

    excerpts = pd.DataFrame()
    if papers is not None:
        source = _as_literature_source(papers, topic=topic)
        if isinstance(source, HypothesisRegistry):
            rows = []
            for item in source.hypotheses:
                rows.append(
                    {
                        "paper_id": None,
                        "page": None,
                        "similarity": _text_similarity(query, item.statement),
                        "text": item.statement,
                        "hypothesis_id": item.hypothesis_id,
                    }
                )
            excerpts = pd.DataFrame(rows).sort_values("similarity", ascending=False).head(top_k) if rows else pd.DataFrame()
        else:
            hits = source.search(query, top_k=top_k)
            excerpts = pd.DataFrame(
                [
                    {
                        "paper_id": hit.paper_id,
                        "page": hit.page,
                        "similarity": _text_similarity(query, hit.text),
                        "text": hit.text,
                        "section": hit.section,
                    }
                    for hit in hits
                ]
            )
            if len(excerpts):
                excerpts = excerpts.sort_values("similarity", ascending=False).reset_index(drop=True)
    return HypothesisSearchResult(str(query), hypothesis_matches, excerpts)

discover

discover(*, data: DataFrame | Series | Mapping[str, Any] | None = None, papers: LiteratureCorpus | HypothesisRegistry | str | Path | Sequence[Any] | None = None, predictions: DataFrame | Mapping[str, Sequence[float]] | None = None, robustness_results: DataFrame | None = None, robustness_metric: str | None = None, domain: str = 'quantitative_finance', max_candidates: int = 50, audit_data_candidates: bool = True, **data_kwargs: Any) -> HypothesisCollection

Combine data-driven and literature-driven hypothesis discovery.

Data evidence and novelty evidence remain separate. When both data and a literature corpus are supplied, data-generated candidates are audited against the supplied corpus; no global novelty claim is made automatically.

Source code in src/asrquant/hypotheses.py

def discover(
    *,
    data: pd.DataFrame | pd.Series | Mapping[str, Any] | None = None,
    papers: LiteratureCorpus | HypothesisRegistry | str | Path | Sequence[Any] | None = None,
    predictions: pd.DataFrame | Mapping[str, Sequence[float]] | None = None,
    robustness_results: pd.DataFrame | None = None,
    robustness_metric: str | None = None,
    domain: str = "quantitative_finance",
    max_candidates: int = 50,
    audit_data_candidates: bool = True,
    **data_kwargs: Any,
) -> HypothesisCollection:
    """Combine data-driven and literature-driven hypothesis discovery.

    Data evidence and novelty evidence remain separate.  When both data and a
    literature corpus are supplied, data-generated candidates are audited against
    the supplied corpus; no global novelty claim is made automatically.
    """
    collections: list[HypothesisCollection] = []
    active_domain = _normalise_domain(domain)
    if data is not None:
        collections.append(from_data(data, domain=active_domain, max_candidates=max_candidates, **data_kwargs))
    if papers is not None:
        collections.append(from_literature(papers, topic=active_domain, max_candidates=max_candidates))
    if predictions is not None:
        collections.append(from_model_disagreement(predictions, domain=active_domain, max_candidates=max_candidates))
    if robustness_results is not None:
        if robustness_metric is None:
            raise InputValidationError("robustness_metric is required with robustness_results")
        collections.append(
            from_robustness(
                robustness_results,
                metric=robustness_metric,
                domain=active_domain,
                max_candidates=max_candidates,
            )
        )
    if not collections:
        raise InputValidationError("provide at least one of data, papers, predictions, or robustness_results")

    items = [item for collection in collections for item in collection.hypotheses]
    if papers is not None and data is not None and audit_data_candidates:
        audited: list[HypothesisIdea] = []
        for item in items:
            if item.source.startswith("data"):
                result = audit(item, corpus=papers, topic=active_domain, top_k=5)
                references = result.closest_matches.to_dict(orient="records")
                audited.append(
                    replace(
                        item,
                        novelty_status=result.novelty_status,
                        references=item.references + references,
                        metadata={
                            **item.metadata,
                            "novelty_audit_recommendation": result.recommendation,
                            "corpus_fingerprint": result.corpus_fingerprint,
                        },
                    )
                )
            else:
                audited.append(item)
        items = audited

    items = _deduplicate(items)
    items = sorted(items, key=lambda item: item.priority_score, reverse=True)[:max_candidates]
    metadata = {
        "source": "combined" if len(collections) > 1 else collections[0].metadata.get("source", "unknown"),
        "domain": active_domain,
        "component_sources": [collection.metadata.get("source") for collection in collections],
        "tests_performed": int(sum(int(collection.metadata.get("tests_performed", 0)) for collection in collections)),
        "multiple_testing": "Benjamini-Hochberg FDR" if any(collection.metadata.get("tests_performed", 0) for collection in collections) else "not_applicable",
        "novelty_rule": "Never established automatically; corpus-relative prior-art audit only.",
    }
    return HypothesisCollection(items, metadata)