diff --git a/.github/workflows/cross_repo_smoke.yml b/.github/workflows/cross_repo_smoke.yml index 87f7693..40362f5 100644 --- a/.github/workflows/cross_repo_smoke.yml +++ b/.github/workflows/cross_repo_smoke.yml @@ -4,7 +4,7 @@ on: workflow_dispatch: inputs: as_of: - description: "Smoke date. Defaults to current UTC date when empty." + description: "Smoke date. Defaults to the most recently closed UTC date when empty." required: false type: string schedule: @@ -48,7 +48,11 @@ jobs: INPUT_AS_OF: ${{ github.event.inputs.as_of || '' }} run: | set -euo pipefail - AS_OF="${INPUT_AS_OF:-$(date -u +%F)}" + if [ -n "${INPUT_AS_OF}" ]; then + AS_OF="${INPUT_AS_OF}" + else + AS_OF="$(python -c 'from quant_advisor_research.build_pipeline import default_daily_as_of; print(default_daily_as_of().isoformat())')" + fi python scripts/run_cross_repo_smoke.py \ --as-of "${AS_OF}" \ --political-events ../political-events/data/live/political_events.csv \ diff --git a/.github/workflows/monthly_advisory_review.yml b/.github/workflows/monthly_advisory_review.yml index 6a7b790..9266d92 100644 --- a/.github/workflows/monthly_advisory_review.yml +++ b/.github/workflows/monthly_advisory_review.yml @@ -4,7 +4,7 @@ on: workflow_dispatch: inputs: as_of: - description: "Report date. Defaults to current UTC date when empty." + description: "Report date. Defaults to the last day of the most recently closed UTC month when empty." required: false type: string previous_report_path: @@ -104,7 +104,11 @@ jobs: MARKET_DATA_PROXY_POOL_URL: ${{ github.event.inputs.market_data_proxy_pool_url || vars.MARKET_DATA_PROXY_POOL_URL || '' }} run: | set -euo pipefail - AS_OF="${INPUT_AS_OF:-$(date -u +%F)}" + if [ -n "${INPUT_AS_OF}" ]; then + AS_OF="${INPUT_AS_OF}" + else + AS_OF="$(python -c 'from quant_advisor_research.build_pipeline import default_monthly_as_of; print(default_monthly_as_of().isoformat())')" + fi mkdir -p .cache/market-data ARGS=( --as-of "${AS_OF}" diff --git a/.github/workflows/publish_advisory_site.yml b/.github/workflows/publish_advisory_site.yml index 5356d50..5e5bf01 100644 --- a/.github/workflows/publish_advisory_site.yml +++ b/.github/workflows/publish_advisory_site.yml @@ -4,7 +4,7 @@ on: workflow_dispatch: inputs: as_of: - description: "Report date. Defaults to the most recent Saturday when empty." + description: "Report date. Defaults to the most recent fully closed Saturday when empty." required: false type: string site_url: diff --git a/.github/workflows/weekly_advisory_review.yml b/.github/workflows/weekly_advisory_review.yml index 62f0041..416823a 100644 --- a/.github/workflows/weekly_advisory_review.yml +++ b/.github/workflows/weekly_advisory_review.yml @@ -4,7 +4,7 @@ on: workflow_dispatch: inputs: as_of: - description: "Report date. Defaults to current UTC date when empty." + description: "Report date. Defaults to the most recent fully closed Saturday when empty." required: false type: string political_events_path: @@ -98,7 +98,11 @@ jobs: MARKET_DATA_PROXY_POOL_URL: ${{ github.event.inputs.market_data_proxy_pool_url || vars.MARKET_DATA_PROXY_POOL_URL || '' }} run: | set -euo pipefail - AS_OF="${INPUT_AS_OF:-$(date -u +%F)}" + if [ -n "${INPUT_AS_OF}" ]; then + AS_OF="${INPUT_AS_OF}" + else + AS_OF="$(python -c 'from quant_advisor_research.build_pipeline import default_weekly_as_of; print(default_weekly_as_of().isoformat())')" + fi mkdir -p .cache/market-data ARGS=( --as-of "${AS_OF}" diff --git a/docs/advisory_contract.md b/docs/advisory_contract.md index 159def7..9f43fe1 100644 --- a/docs/advisory_contract.md +++ b/docs/advisory_contract.md @@ -5,9 +5,15 @@ Required top-level fields: ```text -schema_version: "5" +schema_version: "6" +contract_version: "model_recommendations.v6" as_of: ISO date +reference_time: exclusive start of the UTC day after as_of generated_at: ISO datetime +expires_at: generated_at + 7 days +input_digest: 64 lowercase hexadecimal SHA-256 characters +freshness.ai_signal +freshness.theme_momentum mode: "model_recommendations" cadence: "daily" | "weekly" | "monthly" audience_scope: "non_personalized_model_research" @@ -19,12 +25,10 @@ final_decisions: object, optional policy: object ``` -The current CLI and scheduled workflows remain explicit v5 producers. Readers -accept both v5 and v6, but a producer must opt in to v6 only after it can emit -all v6 provenance fields. No v6 field is silently added to, or accepted on, a -v5 report. +The current CLI and scheduled workflows produce v6. Readers continue to accept +legacy v5 reports, but no v6 field is silently accepted on a v5 report. -### V6 dual-read migration contract +### V6 default contract A v6 report uses `contract_version = model_recommendations.v6` and adds these required top-level fields: @@ -307,9 +311,9 @@ Required manifest fields: ```text manifest_type = model_recommendation_report artifact_type = model_recommendations -contract_version = model_recommendations.v5 -schema_version = 5 -version = --schema-5- +contract_version = model_recommendations.v6 +schema_version = 6 +version = --schema-6- source_project = QuantAdvisorResearch producer.repository producer.git_sha @@ -320,10 +324,11 @@ artifacts.json.sha256 artifacts.markdown.sha256 policy generated_at +input_digest ``` -For a v6 report, the manifest additionally carries the exact top-level -`input_digest`. V5 manifests remain unchanged and omit this field. +The v6 manifest carries the exact top-level `input_digest`. Legacy v5 manifests +remain readable and omit this field. ## Source Mode diff --git a/examples/research_signal_context.example.json b/examples/research_signal_context.example.json index a663c24..aee903a 100644 --- a/examples/research_signal_context.example.json +++ b/examples/research_signal_context.example.json @@ -1,5 +1,7 @@ { - "schema_version": "1", + "schema_version": "2", + "model_version": "shadow-v2", + "scoring_version": "rules-v2", "as_of": "2026-05-28", "generated_at": "2026-05-28T14:38:47.474891Z", "mode": "shadow", @@ -19,7 +21,7 @@ "semiconductor_and_leveraged_etf_high_volatility", "no_point_in_time_replay_evidence" ], - "research_bias": { + "candidate_bias": { "IDX1": "positive", "IDX2": "positive", "SECT1": "watch", diff --git a/scripts/d3_build_daily_preview.py b/scripts/d3_build_daily_preview.py index cb29fb8..2caa46d 100644 --- a/scripts/d3_build_daily_preview.py +++ b/scripts/d3_build_daily_preview.py @@ -12,7 +12,7 @@ from pathlib import Path from unittest.mock import patch -from d3_evidence import DEPENDENCY_INVENTORY, EVIDENCE_VERSION, BundleSnapshot, EvidenceContractError, locked_environment_evidence, repository_file_hashes, validate_exact_bundle, validate_preview_snapshot +from d3_evidence import DEPENDENCY_INVENTORY, EVIDENCE_VERSION, BundleSnapshot, EvidenceContractError, locked_environment_evidence, preview_source_contract, repository_file_hashes, validate_exact_bundle, validate_preview_snapshot from quant_advisor_research.advisory_report import build_advisory_report from quant_advisor_research.preview_workspace import build_preview_workspace @@ -45,13 +45,14 @@ def build_evidence(args: argparse.Namespace) -> None: raise EvidenceContractError("repeat_workspace_not_distinct") first_snapshot = validate_exact_bundle(first) second_snapshot = validate_exact_bundle(second) - validate_preview_snapshot(first_snapshot); validate_preview_snapshot(second_snapshot) + validate_preview_snapshot(first_snapshot) + validate_preview_snapshot(second_snapshot) first_files, second_files = _bundle_hashes(first_snapshot), _bundle_hashes(second_snapshot) if first_files != second_files or any(first_snapshot.member(name).content != second_snapshot.member(name).content for name in ("manifest.json", "report.html", "report.json")): raise EvidenceContractError("repeat_build_not_equal") manifest = json.loads(first_snapshot.member("manifest.json").content.decode("utf-8")) - source = {"schema_version": "5", "contract_version": "model_recommendations.v5", "cadence": "daily", "as_of": args.as_of, "generated_at": args.frozen_generated_at} - if manifest.get("bundle_contract") != "qar.preview_bundle.v1" or manifest.get("source") != source: + source = preview_source_contract(manifest, as_of=args.as_of, generated_at=args.frozen_generated_at) + if manifest.get("bundle_contract") != "qar.preview_bundle.v1": raise EvidenceContractError("source_contract_mismatch") evidence = { "evidence_version": EVIDENCE_VERSION, "base_sha": args.base_sha, "head_sha": args.head_sha, diff --git a/scripts/d3_evidence.py b/scripts/d3_evidence.py index 3378184..591a7ab 100644 --- a/scripts/d3_evidence.py +++ b/scripts/d3_evidence.py @@ -23,6 +23,7 @@ "src/quant_advisor_research/time_contract.py", "tests/test_d3_exact_bundle.py", "examples/political_events.example.csv", "examples/political_watchlist.example.csv", ] +SOURCE_CONTRACTS = {"5": "model_recommendations.v5", "6": "model_recommendations.v6"} class EvidenceContractError(ValueError): @@ -49,6 +50,33 @@ def member(self, name: str) -> BundleMemberSnapshot: raise EvidenceContractError("bundle_member_set_invalid") +def preview_source_contract( + manifest: Mapping[str, object], + *, + as_of: str, + generated_at: str, +) -> dict[str, object]: + source = manifest.get("source") + if not isinstance(source, Mapping) or set(source) != { + "schema_version", + "contract_version", + "cadence", + "as_of", + "generated_at", + }: + raise EvidenceContractError("source_contract_mismatch") + schema_version = source.get("schema_version") + if ( + type(schema_version) is not str + or source.get("contract_version") != SOURCE_CONTRACTS.get(schema_version) + or source.get("cadence") != "daily" + or source.get("as_of") != as_of + or source.get("generated_at") != generated_at + ): + raise EvidenceContractError("source_contract_mismatch") + return dict(source) + + def _directory_flags() -> int: required = ("O_DIRECTORY", "O_CLOEXEC", "O_NOFOLLOW") if any(not hasattr(os, name) for name in required): diff --git a/scripts/d3_verify_daily_preview.py b/scripts/d3_verify_daily_preview.py index 5886623..6f7b754 100644 --- a/scripts/d3_verify_daily_preview.py +++ b/scripts/d3_verify_daily_preview.py @@ -9,7 +9,7 @@ import platform from pathlib import Path -from d3_evidence import DEPENDENCY_INVENTORY, EVIDENCE_VERSION, BundleSnapshot, EvidenceContractError, locked_environment_evidence, repository_file_hashes, validate_exact_bundle, validate_preview_snapshot +from d3_evidence import DEPENDENCY_INVENTORY, EVIDENCE_VERSION, BundleSnapshot, EvidenceContractError, locked_environment_evidence, preview_source_contract, repository_file_hashes, validate_exact_bundle, validate_preview_snapshot EXPECTED_KEYS = frozenset({"evidence_version", "base_sha", "head_sha", "source", "deterministic_clock", "workflow_dependency_inventory", "dependency_files", "locked_environment", "bundle", "repeat_build"}) @@ -31,10 +31,14 @@ def verify(args: argparse.Namespace) -> None: if evidence["workflow_dependency_inventory"] != DEPENDENCY_INVENTORY or evidence["dependency_files"] != dependency_files: raise EvidenceContractError("dependency_evidence_mismatch") environment = locked_environment_evidence(lock_sha256=hashlib.sha256(Path(args.lock_path).read_bytes()).hexdigest(), uv_version=args.uv_version, python_version=platform.python_version(), distributions=_distributions()) - expected_source = {"schema_version": "5", "contract_version": "model_recommendations.v5", "cadence": "daily", "as_of": args.as_of, "generated_at": args.frozen_generated_at} snapshot = validate_exact_bundle(args.workspace) _, manifest = validate_preview_snapshot(snapshot) - if manifest.get("bundle_contract") != "qar.preview_bundle.v1" or manifest.get("source") != expected_source: + expected_source = preview_source_contract( + manifest, + as_of=args.as_of, + generated_at=args.frozen_generated_at, + ) + if manifest.get("bundle_contract") != "qar.preview_bundle.v1": raise EvidenceContractError("source_contract_mismatch") files = _hashes(snapshot) expected = { diff --git a/src/quant_advisor_research/advisory_report.py b/src/quant_advisor_research/advisory_report.py index 767973b..a8910ac 100644 --- a/src/quant_advisor_research/advisory_report.py +++ b/src/quant_advisor_research/advisory_report.py @@ -5,13 +5,22 @@ import json import os from collections import defaultdict +from collections.abc import Mapping from dataclasses import dataclass from pathlib import Path from typing import Any -from .artifacts import write_report_manifest +from .artifacts import input_digest_for_payloads, write_report_manifest from .contracts import ALLOWED_CADENCES, validate_advisory_report -from .csv_utils import read_csv_rows +from .csv_utils import read_csv_rows, read_csv_rows_bytes +from .time_contract import ( + REPORT_EXPIRY_DAYS, + ContextFreshness, + assess_context_freshness, + contract_version_for_schema, + normalize_aware_datetime, + report_time_bounds, +) EVENT_WEIGHTS = { @@ -47,6 +56,48 @@ "negative": -3, } +AI_LONG_CONTEXT_SCORES = { + "positive": 0.5, + "watch": 0.35, +} + +AI_SIGNAL_REQUIRED_KEYS = frozenset( + { + "schema_version", + "as_of", + "generated_at", + "mode", + "horizon", + "universe", + "regime", + "risk_flags", + "candidate_bias", + "confidence", + "evidence", + "expires_at", + "policy", + } +) +AI_SIGNAL_ALLOWED_KEYS = AI_SIGNAL_REQUIRED_KEYS | frozenset( + { + "model_version", + "scoring_version", + "theme_bias", + "symbol_bias", + "symbol_theme_exposure", + } +) +AI_SIGNAL_SCHEMA_VERSIONS = frozenset({"1", "2"}) +AI_SIGNAL_REGIMES = frozenset({"risk_on", "risk_off", "neutral", "mixed", "unknown"}) +AI_SIGNAL_BIASES = frozenset({"positive", "negative", "neutral", "watch", "avoid"}) +AI_SIGNAL_HORIZON = "1-3 years" +AI_BIAS_ALLOWED_KEYS = frozenset({"bias", "confidence", "rationale", "horizon", "risk_flags", "linked_themes"}) +AI_EVIDENCE_ALLOWED_KEYS = frozenset({"sources", "summary", "data_gaps"}) +AI_POLICY_ALLOWED_KEYS = frozenset({"execution_allowed", "portfolio_allocation_allowed", "downstream_use"}) +AI_POLICY_FORBIDDEN_TERMS = frozenset({"live", "allocation", "broker", "execution", "order", "position", "account"}) +AI_POLICY_BLOCKING_TERMS = frozenset({"blocked", "not allowed", "do not", "never", "no "}) +_UNAVAILABLE_INPUT = object() + HORIZON_WINDOWS = { "short": "1-10个交易日", "medium": "2-12周", @@ -214,6 +265,10 @@ class MarketConfirmation: price_observation_count: int = 0 +class AISignalValidationError(ValueError): + """Raised with a sanitized reason when AI context is not consumable.""" + + def parse_date(value: str) -> dt.date: return dt.date.fromisoformat(value.strip()) @@ -222,9 +277,156 @@ def utc_now_iso() -> str: return dt.datetime.now(dt.UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") -def load_events(path: str | Path, as_of: dt.date) -> list[Event]: +def utc_iso(value: dt.datetime) -> str: + return value.astimezone(dt.UTC).isoformat().replace("+00:00", "Z") + + +def _ai_contract_invalid() -> None: + raise AISignalValidationError("ai_signal_contract_invalid") + + +def _valid_ai_string(value: Any) -> bool: + return isinstance(value, str) and bool(value.strip()) + + +def _valid_ai_string_list(value: Any, *, allow_empty: bool = False) -> bool: + return ( + isinstance(value, list) + and (allow_empty or bool(value)) + and all(_valid_ai_string(item) for item in value) + ) + + +def _valid_ai_confidence(value: Any) -> bool: + return isinstance(value, (int, float)) and not isinstance(value, bool) and 0 <= value <= 1 + + +def _validate_ai_bias_value(value: Any) -> None: + if isinstance(value, str): + bias = value + elif isinstance(value, Mapping): + if set(value) - AI_BIAS_ALLOWED_KEYS: + _ai_contract_invalid() + bias = value.get("bias") + if not _valid_ai_string(bias): + _ai_contract_invalid() + if "confidence" in value and not _valid_ai_confidence(value["confidence"]): + _ai_contract_invalid() + if any(key in value and not _valid_ai_string(value[key]) for key in ("rationale", "horizon")): + _ai_contract_invalid() + if any( + key in value and not _valid_ai_string_list(value[key], allow_empty=True) + for key in ("risk_flags", "linked_themes") + ): + _ai_contract_invalid() + else: + _ai_contract_invalid() + if not _valid_ai_string(bias) or bias not in AI_SIGNAL_BIASES: + _ai_contract_invalid() + + +def _validate_ai_bias_mapping(value: Any) -> None: + if not isinstance(value, Mapping): + _ai_contract_invalid() + for key, bias in value.items(): + if not _valid_ai_string(key): + _ai_contract_invalid() + _validate_ai_bias_value(bias) + + +def validate_ai_signal(payload: Any) -> None: + if not isinstance(payload, Mapping) or not AI_SIGNAL_REQUIRED_KEYS <= set(payload): + _ai_contract_invalid() + if set(payload) - AI_SIGNAL_ALLOWED_KEYS: + _ai_contract_invalid() + schema_version = payload["schema_version"] + if not _valid_ai_string(schema_version) or schema_version not in AI_SIGNAL_SCHEMA_VERSIONS: + _ai_contract_invalid() + if schema_version == "2" and any( + not _valid_ai_string(payload.get(key)) for key in ("model_version", "scoring_version") + ): + _ai_contract_invalid() + try: + if not _valid_ai_string(payload["as_of"]): + _ai_contract_invalid() + dt.date.fromisoformat(payload["as_of"]) + if not _valid_ai_string(payload["generated_at"]): + _ai_contract_invalid() + normalize_aware_datetime(payload["generated_at"]) + if not _valid_ai_string(payload["expires_at"]): + _ai_contract_invalid() + dt.date.fromisoformat(payload["expires_at"]) + except (TypeError, ValueError): + _ai_contract_invalid() + if payload["mode"] != "shadow" or payload["horizon"] != AI_SIGNAL_HORIZON: + _ai_contract_invalid() + if not _valid_ai_string(payload["regime"]) or payload["regime"] not in AI_SIGNAL_REGIMES: + _ai_contract_invalid() + if not _valid_ai_string_list(payload["universe"]): + _ai_contract_invalid() + if not _valid_ai_string_list(payload["risk_flags"], allow_empty=True): + _ai_contract_invalid() + _validate_ai_bias_mapping(payload["candidate_bias"]) + for key in ("theme_bias", "symbol_bias"): + if key in payload: + _validate_ai_bias_mapping(payload[key]) + if "symbol_theme_exposure" in payload: + exposure = payload["symbol_theme_exposure"] + if not isinstance(exposure, Mapping): + _ai_contract_invalid() + for symbol, theme_ids in exposure.items(): + if not _valid_ai_string(symbol) or not _valid_ai_string_list(theme_ids): + _ai_contract_invalid() + if not _valid_ai_confidence(payload["confidence"]): + _ai_contract_invalid() + evidence = payload["evidence"] + if not isinstance(evidence, Mapping) or set(evidence) - AI_EVIDENCE_ALLOWED_KEYS: + _ai_contract_invalid() + if not _valid_ai_string_list(evidence.get("sources")): + _ai_contract_invalid() + if not _valid_ai_string(evidence.get("summary")): + _ai_contract_invalid() + if not _valid_ai_string_list(evidence.get("data_gaps", []), allow_empty=True): + _ai_contract_invalid() + policy = payload["policy"] + if not isinstance(policy, Mapping) or set(policy) - AI_POLICY_ALLOWED_KEYS: + _ai_contract_invalid() + downstream_use = policy.get("downstream_use") + if ( + policy.get("execution_allowed") is not False + or policy.get("portfolio_allocation_allowed", False) is not False + or not _valid_ai_string(downstream_use) + ): + _ai_contract_invalid() + normalized_use = str(downstream_use).lower() + if not any(term in normalized_use for term in ("research", "advisory", "shadow")): + _ai_contract_invalid() + if "live" in normalized_use or ( + any(term in normalized_use for term in AI_POLICY_FORBIDDEN_TERMS) + and not any(term in normalized_use for term in AI_POLICY_BLOCKING_TERMS) + ): + _ai_contract_invalid() + + +def freshness_record(result: ContextFreshness, payload: Mapping[str, Any] | None) -> dict[str, Any]: + record: dict[str, Any] = { + "present": result.present, + "valid": result.valid, + "reason": result.reason, + } + if result.present and payload is not None: + for key in ("as_of", "generated_at", "expires_at"): + if key in payload: + record[key] = payload[key] + if result.reason == "legacy_expiry_compatibility": + record["compatibility_warning"] = "missing_expires_at" + return record + + +def load_events(path: str | Path, as_of: dt.date, *, source_bytes: bytes | None = None) -> list[Event]: events: list[Event] = [] - for row in read_csv_rows(path): + rows = read_csv_rows_bytes(source_bytes) if source_bytes is not None else read_csv_rows(path) + for row in rows: event_date = parse_date(row["event_date"]) if event_date > as_of: continue @@ -270,9 +472,10 @@ def entity_evidence_details(events: list[Event]) -> list[dict[str, Any]]: ] -def load_watchlist(path: str | Path) -> dict[str, WatchlistItem]: +def load_watchlist(path: str | Path, *, source_bytes: bytes | None = None) -> dict[str, WatchlistItem]: items: dict[str, WatchlistItem] = {} - for row in read_csv_rows(path): + rows = read_csv_rows_bytes(source_bytes) if source_bytes is not None else read_csv_rows(path) + for row in rows: symbol = row["symbol"].upper() items[symbol] = WatchlistItem( symbol=symbol, @@ -285,24 +488,24 @@ def load_watchlist(path: str | Path) -> dict[str, WatchlistItem]: return items -def load_ai_signal(path: str | Path | None) -> dict[str, Any] | None: +def load_ai_signal(path: str | Path | None, *, source_bytes: bytes | None = None) -> dict[str, Any] | None: if path is None: return None - with Path(path).open(encoding="utf-8") as handle: - payload = json.load(handle) - if payload.get("mode") != "shadow": - raise ValueError("AI signal input must remain mode=shadow.") - if payload.get("policy", {}).get("execution_allowed") is not False: - raise ValueError("AI signal input must not allow execution.") + try: + payload = json.loads(source_bytes.decode("utf-8")) if source_bytes is not None else json.loads(Path(path).read_text(encoding="utf-8")) + except (json.JSONDecodeError, UnicodeError): + raise AISignalValidationError("ai_signal_invalid_json") from None + except OSError: + raise AISignalValidationError("ai_signal_unavailable") from None + validate_ai_signal(payload) return payload -def load_theme_momentum(path: str | Path | None) -> dict[str, Any] | None: +def load_theme_momentum(path: str | Path | None, *, source_bytes: bytes | None = None) -> dict[str, Any] | None: if path is None: return None - with Path(path).open(encoding="utf-8") as handle: - payload = json.load(handle) + payload = json.loads(source_bytes.decode("utf-8")) if source_bytes is not None else json.loads(Path(path).read_text(encoding="utf-8")) if payload.get("mode") != "theme_momentum_snapshot": raise ValueError("Theme momentum input must remain mode=theme_momentum_snapshot.") if payload.get("policy", {}).get("execution_allowed") is not False: @@ -368,11 +571,17 @@ def optional_date(value: Any) -> dt.date | None: return parse_date(text) -def load_market_confirmation(path: str | Path | None, as_of: dt.date) -> dict[str, MarketConfirmation]: +def load_market_confirmation( + path: str | Path | None, + as_of: dt.date, + *, + source_bytes: bytes | None = None, +) -> dict[str, MarketConfirmation]: if path is None: return {} confirmations: dict[str, MarketConfirmation] = {} - for row in read_csv_rows(path): + rows = read_csv_rows_bytes(source_bytes) if source_bytes is not None else read_csv_rows(path) + for row in rows: symbol = str(row.get("symbol", "")).upper().strip() if not symbol: continue @@ -493,7 +702,7 @@ def resolve_ai_bias(symbol: str, ai_signal: dict[str, Any] | None) -> tuple[str return None, [], None normalized_symbol = symbol.upper() explicit_bias: dict[str, Any] = {} - for key in ("candidate_bias", "research_bias", "symbol_bias"): + for key in ("candidate_bias", "symbol_bias"): explicit_bias.update(normalize_ai_mapping(ai_signal.get(key) or {})) if normalized_symbol in explicit_bias: raw_value = explicit_bias[normalized_symbol] @@ -784,8 +993,6 @@ def build_recommendation( evidence_score += AI_BIAS_WEIGHTS.get(ai_bias, 0) if ai_bias in {"avoid", "negative"}: risk_score += 4 - if ai_bias == "positive": - evidence_score += round(ai_confidence * 2) risk_flags = list(ai_signal.get("risk_flags", [])) if ai_signal else [] if any("volatility" in flag or "high_vol" in flag for flag in risk_flags): @@ -858,9 +1065,7 @@ def build_recommendation( ] ) - long_horizon_ai_score = 0.0 - if ai_bias in {"positive", "watch", "neutral"}: - long_horizon_ai_score = round(clamp(ai_confidence, 0, 1), 3) + long_horizon_ai_score = AI_LONG_CONTEXT_SCORES.get(ai_bias or "", 0.0) return { "symbol": symbol, @@ -1460,6 +1665,17 @@ def infer_long_context_missing_reason(ai_signal: dict[str, Any] | None) -> str: return "current_candidates_do_not_meet_long_context_gate" +def _snapshot_input(path: str | Path | None, *, fail_on_unavailable: bool) -> bytes | object | None: + if path is None: + return None + try: + return Path(path).read_bytes() + except OSError: + if fail_on_unavailable: + raise + return _UNAVAILABLE_INPUT + + def build_advisory_report( *, as_of: str, @@ -1474,11 +1690,119 @@ def build_advisory_report( if cadence not in ALLOWED_CADENCES: raise ValueError(f"cadence must be one of: {', '.join(sorted(ALLOWED_CADENCES))}") as_of_date = parse_date(as_of) - watchlist = load_watchlist(political_watchlist_path) - events = load_events(political_events_path, as_of_date) - ai_signal = load_ai_signal(ai_signal_path) - theme_momentum = load_theme_momentum(theme_momentum_path) - market_confirmations = load_market_confirmation(market_confirmation_path, as_of_date) + report_bounds = report_time_bounds(as_of_date, utc_now_iso()) + input_payloads = { + "political_events": _snapshot_input(political_events_path, fail_on_unavailable=True), + "political_watchlist": _snapshot_input(political_watchlist_path, fail_on_unavailable=True), + "ai_signal": _snapshot_input(ai_signal_path, fail_on_unavailable=False), + "theme_momentum": _snapshot_input(theme_momentum_path, fail_on_unavailable=False), + "market_confirmation": _snapshot_input(market_confirmation_path, fail_on_unavailable=True), + } + events_bytes = input_payloads["political_events"] + watchlist_bytes = input_payloads["political_watchlist"] + if not isinstance(events_bytes, bytes) or not isinstance(watchlist_bytes, bytes): + raise ValueError("required advisory input unavailable") + watchlist = load_watchlist(political_watchlist_path, source_bytes=watchlist_bytes) + events = load_events(political_events_path, as_of_date, source_bytes=events_bytes) + ai_signal: dict[str, Any] | None = None + ai_source_artifact = "" + ai_freshness_result = assess_context_freshness( + None, + report_as_of=as_of_date, + reference_time=report_bounds.reference_time, + report_generated_at=report_bounds.generated_at, + max_age_days=REPORT_EXPIRY_DAYS, + ) + ai_freshness_payload: Mapping[str, Any] | None = None + ai_quality_warnings: list[str] = [] + if ai_signal_path: + ai_bytes = input_payloads["ai_signal"] + if ai_bytes is _UNAVAILABLE_INPUT: + ai_quality_warnings.append("ai_signal_unavailable") + else: + try: + candidate_ai_signal = load_ai_signal( + ai_signal_path, + source_bytes=ai_bytes if isinstance(ai_bytes, bytes) else None, + ) + except AISignalValidationError as exc: + ai_quality_warnings.append(str(exc)) + else: + ai_freshness_payload = candidate_ai_signal + ai_freshness_result = assess_context_freshness( + candidate_ai_signal, + report_as_of=as_of_date, + reference_time=report_bounds.reference_time, + report_generated_at=report_bounds.generated_at, + max_age_days=REPORT_EXPIRY_DAYS, + ) + ai_source_artifact = str(ai_signal_path) + if ai_freshness_result.valid: + ai_signal = candidate_ai_signal + else: + ai_quality_warnings.append(f"ai_signal_{ai_freshness_result.reason}") + + theme_momentum: dict[str, Any] | None = None + theme_source_artifact = "" + theme_freshness_result = assess_context_freshness( + None, + report_as_of=as_of_date, + reference_time=report_bounds.reference_time, + report_generated_at=report_bounds.generated_at, + max_age_days=REPORT_EXPIRY_DAYS, + ) + theme_freshness_payload: Mapping[str, Any] | None = None + theme_quality_warnings: list[str] = [] + if theme_momentum_path: + theme_bytes = input_payloads["theme_momentum"] + if theme_bytes is _UNAVAILABLE_INPUT: + theme_quality_warnings.append("theme_momentum_contract_invalid") + else: + try: + candidate_theme_momentum = load_theme_momentum( + theme_momentum_path, + source_bytes=theme_bytes if isinstance(theme_bytes, bytes) else None, + ) + except (OSError, TypeError, ValueError, json.JSONDecodeError): + theme_quality_warnings.append("theme_momentum_contract_invalid") + else: + theme_freshness_payload = candidate_theme_momentum + if not candidate_theme_momentum.get("expires_at"): + theme_freshness_payload = { + **candidate_theme_momentum, + "reason": "legacy_expiry_compatibility", + "compatibility_warning": "missing_expires_at", + } + theme_freshness_result = assess_context_freshness( + theme_freshness_payload, + report_as_of=as_of_date, + reference_time=report_bounds.reference_time, + report_generated_at=report_bounds.generated_at, + max_age_days=REPORT_EXPIRY_DAYS, + allow_legacy_expiry=True, + ) + if theme_freshness_result.valid: + theme_source_artifact = str(theme_momentum_path) + theme_momentum = candidate_theme_momentum + else: + theme_quality_warnings.append(f"theme_momentum_{theme_freshness_result.reason}") + if not candidate_theme_momentum.get("expires_at"): + theme_freshness_payload = None + theme_freshness_result = assess_context_freshness( + None, + report_as_of=as_of_date, + reference_time=report_bounds.reference_time, + report_generated_at=report_bounds.generated_at, + max_age_days=REPORT_EXPIRY_DAYS, + ) + else: + theme_source_artifact = str(theme_momentum_path) + market_bytes = input_payloads["market_confirmation"] + market_confirmations = load_market_confirmation( + market_confirmation_path, + as_of_date, + source_bytes=market_bytes if isinstance(market_bytes, bytes) else None, + ) theme_momentum_summary = summarize_theme_momentum(theme_momentum) source_mode, data_quality_warnings = source_mode_for_paths( political_events_path, @@ -1487,6 +1811,7 @@ def build_advisory_report( theme_momentum_path, market_confirmation_path, ) + data_quality_warnings = dedupe(data_quality_warnings + ai_quality_warnings + theme_quality_warnings) events_by_symbol: dict[str, list[Event]] = defaultdict(list) for event in events: @@ -1495,7 +1820,7 @@ def build_advisory_report( symbols = set(watchlist) | set(events_by_symbol) if ai_signal: symbols |= {symbol.upper() for symbol in ai_signal.get("universe", [])} - for key in ("symbol_bias", "research_bias", "candidate_bias"): + for key in ("symbol_bias", "candidate_bias"): symbols |= {symbol.upper() for symbol in normalize_ai_mapping(ai_signal.get(key) or {})} all_recommendations = [ @@ -1514,18 +1839,30 @@ def build_advisory_report( final_decisions = build_final_decisions(all_recommendations, theme_momentum, market_confirmations) long_context_symbols = long_context_symbols_from_decisions(final_decisions) + digest_payloads = { + name: "unavailable" if payload is _UNAVAILABLE_INPUT else payload + for name, payload in input_payloads.items() + } report = { - "schema_version": "5", + "schema_version": "6", + "contract_version": contract_version_for_schema("6"), "as_of": as_of_date.isoformat(), - "generated_at": utc_now_iso(), + "reference_time": utc_iso(report_bounds.reference_time), + "generated_at": utc_iso(report_bounds.generated_at), + "expires_at": utc_iso(report_bounds.expires_at), + "input_digest": input_digest_for_payloads(digest_payloads), + "freshness": { + "ai_signal": freshness_record(ai_freshness_result, ai_freshness_payload), + "theme_momentum": freshness_record(theme_freshness_result, theme_freshness_payload), + }, "mode": "model_recommendations", "cadence": cadence, "audience_scope": "non_personalized_model_research", "source_artifacts": { "political_events": str(political_events_path), "political_watchlist": str(political_watchlist_path), - "ai_signal": str(ai_signal_path) if ai_signal_path else "", - "theme_momentum": str(theme_momentum_path) if theme_momentum_path else "", + "ai_signal": ai_source_artifact, + "theme_momentum": theme_source_artifact, "market_confirmation": str(market_confirmation_path) if market_confirmation_path else "", }, "summary": { diff --git a/src/quant_advisor_research/artifacts.py b/src/quant_advisor_research/artifacts.py index f06b10d..baa4946 100644 --- a/src/quant_advisor_research/artifacts.py +++ b/src/quant_advisor_research/artifacts.py @@ -41,6 +41,15 @@ def sha256_bytes(payload: bytes) -> str: return hashlib.sha256(payload).hexdigest() +def input_digest_for_payloads(sources: Mapping[str, bytes | str | None]) -> str: + identities = { + name: sha256_bytes(payload) if isinstance(payload, bytes) else payload + for name, payload in sources.items() + } + canonical = json.dumps(identities, sort_keys=True, separators=(",", ":")).encode("utf-8") + return sha256_bytes(canonical) + + def write_json(path: str | Path, payload: Mapping[str, Any]) -> Path: output_path = Path(path) output_path.parent.mkdir(parents=True, exist_ok=True) diff --git a/src/quant_advisor_research/build_pipeline.py b/src/quant_advisor_research/build_pipeline.py index 5dc8f43..39a9d7d 100644 --- a/src/quant_advisor_research/build_pipeline.py +++ b/src/quant_advisor_research/build_pipeline.py @@ -55,12 +55,22 @@ def parse_date(value: str) -> dt.date: return dt.date.fromisoformat(value.strip()) +def default_daily_as_of(today: dt.date | None = None) -> dt.date: + current = today or dt.datetime.now(dt.UTC).date() + return current - dt.timedelta(days=1) + + def default_weekly_as_of(today: dt.date | None = None) -> dt.date: current = today or dt.datetime.now(dt.UTC).date() - days_since_saturday = (current.weekday() - 5) % 7 + days_since_saturday = (current.weekday() - 5) % 7 or 7 return current - dt.timedelta(days=days_since_saturday) +def default_monthly_as_of(today: dt.date | None = None) -> dt.date: + current = today or dt.datetime.now(dt.UTC).date() + return current.replace(day=1) - dt.timedelta(days=1) + + def existing_optional_path(value: str | Path | None) -> Path | None: if value in {None, ""}: return None diff --git a/src/quant_advisor_research/contracts.py b/src/quant_advisor_research/contracts.py index e35f6fa..73594c0 100644 --- a/src/quant_advisor_research/contracts.py +++ b/src/quant_advisor_research/contracts.py @@ -22,7 +22,7 @@ class AdvisoryValidationError(ValueError): ALLOWED_CADENCES = frozenset({"daily", "weekly", "monthly"}) SOURCE_PROJECT = "QuantAdvisorResearch" -REPORT_CONTRACT_VERSION = "model_recommendations.v5" +REPORT_CONTRACT_VERSION = "model_recommendations.v6" ALLOWED_RECOMMENDATION_RATINGS = frozenset({"recommend", "watch", "verify_source", "defer", "monitor"}) ALLOWED_RECOMMENDATION_TIERS = frozenset({"tier_1", "tier_2", "watchlist", "source_check", "defer", "monitor"}) ALLOWED_HORIZONS = frozenset({"short", "medium", "long", "not_applicable"}) diff --git a/src/quant_advisor_research/csv_utils.py b/src/quant_advisor_research/csv_utils.py index ff2587a..927a8fa 100644 --- a/src/quant_advisor_research/csv_utils.py +++ b/src/quant_advisor_research/csv_utils.py @@ -1,6 +1,7 @@ from __future__ import annotations import csv +import io from pathlib import Path from typing import Iterable @@ -10,6 +11,11 @@ def read_csv_rows(path: str | Path) -> list[dict[str, str]]: return [dict(row) for row in csv.DictReader(handle)] +def read_csv_rows_bytes(payload: bytes) -> list[dict[str, str]]: + with io.StringIO(payload.decode("utf-8"), newline="") as handle: + return [dict(row) for row in csv.DictReader(handle)] + + def write_csv_rows(path: str | Path, fieldnames: list[str], rows: Iterable[dict[str, object]]) -> None: output_path = Path(path) output_path.parent.mkdir(parents=True, exist_ok=True) @@ -18,4 +24,3 @@ def write_csv_rows(path: str | Path, fieldnames: list[str], rows: Iterable[dict[ writer.writeheader() for row in rows: writer.writerow({key: row.get(key, "") for key in fieldnames}) - diff --git a/src/quant_advisor_research/preview_bundle.py b/src/quant_advisor_research/preview_bundle.py index e031b19..5d71fc1 100644 --- a/src/quant_advisor_research/preview_bundle.py +++ b/src/quant_advisor_research/preview_bundle.py @@ -14,10 +14,12 @@ from .artifact_integrity import ArtifactIntegrityError, snapshot_json_wire from .contracts import AdvisoryValidationError, validate_advisory_report +from .time_contract import contract_version_for_schema BUNDLE_CONTRACT = "qar.preview_bundle.v1" -SOURCE_SCHEMA_VERSION = "5" -SOURCE_CONTRACT_VERSION = "model_recommendations.v5" +SOURCE_SCHEMA_VERSION = "6" +SOURCE_SCHEMA_VERSIONS = frozenset({"5", "6"}) +SOURCE_CONTRACT_VERSION = "model_recommendations.v6" _FIXED_FILES = frozenset({"report.json", "report.html", "manifest.json"}) _MANIFEST_KEYS = frozenset({"bundle_contract", "source", "artifacts"}) _SOURCE_KEYS = frozenset({"schema_version", "contract_version", "cadence", "as_of", "generated_at"}) @@ -63,9 +65,10 @@ def _validated_source(report: Mapping[str, Any]) -> dict[str, object]: validate_advisory_report(snapshot) except (ArtifactIntegrityError, AdvisoryValidationError, TypeError, ValueError, OverflowError, UnicodeError, RecursionError): raise _error("source_invalid") from None - if snapshot.get("schema_version") != SOURCE_SCHEMA_VERSION: + schema_version = snapshot.get("schema_version") + if schema_version not in SOURCE_SCHEMA_VERSIONS: raise _error("source_schema_unsupported") - if "contract_version" in snapshot: + if schema_version == "5" and "contract_version" in snapshot: raise _error("source_contract_field_forbidden") if snapshot.get("cadence") != "daily": raise _error("daily_only") @@ -93,8 +96,8 @@ def artifact(name: str, role: str, content: bytes) -> dict[str, str]: return { "bundle_contract": BUNDLE_CONTRACT, "source": { - "schema_version": SOURCE_SCHEMA_VERSION, - "contract_version": SOURCE_CONTRACT_VERSION, + "schema_version": snapshot["schema_version"], + "contract_version": contract_version_for_schema(str(snapshot["schema_version"])), "cadence": "daily", "as_of": snapshot["as_of"], "generated_at": snapshot["generated_at"], @@ -197,6 +200,6 @@ def read_preview_bundle(output_dir: str | Path) -> PreviewBundleEvidence: __all__ = [ - "BUNDLE_CONTRACT", "SOURCE_CONTRACT_VERSION", "PreviewBundleError", "PreviewBundleEvidence", + "BUNDLE_CONTRACT", "SOURCE_CONTRACT_VERSION", "SOURCE_SCHEMA_VERSIONS", "PreviewBundleError", "PreviewBundleEvidence", "build_preview_bundle", "read_preview_bundle", ] diff --git a/src/quant_advisor_research/preview_workspace.py b/src/quant_advisor_research/preview_workspace.py index d2fe9aa..6182d54 100644 --- a/src/quant_advisor_research/preview_workspace.py +++ b/src/quant_advisor_research/preview_workspace.py @@ -17,7 +17,8 @@ from .artifact_integrity import ArtifactIntegrityError, snapshot_json_wire from .contracts import AdvisoryValidationError, validate_advisory_report -from .preview_bundle import BUNDLE_CONTRACT, PreviewBundleError, SOURCE_CONTRACT_VERSION, read_preview_bundle +from .preview_bundle import BUNDLE_CONTRACT, PreviewBundleError, SOURCE_SCHEMA_VERSIONS, read_preview_bundle +from .time_contract import contract_version_for_schema _FIXED_FILES = frozenset({"report.json", "report.html", "manifest.json"}) _NOFOLLOW = getattr(os, "O_NOFOLLOW", None) @@ -58,9 +59,10 @@ def _validated_source(report: Mapping[str, Any]) -> dict[str, object]: validate_advisory_report(snapshot) except (ArtifactIntegrityError, AdvisoryValidationError, TypeError, ValueError, OverflowError, UnicodeError, RecursionError): raise _error("source_invalid") from None - if snapshot.get("schema_version") != "5": + schema_version = snapshot.get("schema_version") + if schema_version not in SOURCE_SCHEMA_VERSIONS: raise _error("source_schema_unsupported") - if "contract_version" in snapshot or snapshot.get("cadence") != "daily": + if (schema_version == "5" and "contract_version" in snapshot) or snapshot.get("cadence") != "daily": raise _error("source_contract_invalid") if type(snapshot.get("as_of")) is not str or type(snapshot.get("generated_at")) is not str: raise _error("source_time_invalid") @@ -93,8 +95,8 @@ def artifact(name: str, role: str, content: bytes) -> dict[str, str]: return { "bundle_contract": BUNDLE_CONTRACT, "source": { - "schema_version": "5", - "contract_version": SOURCE_CONTRACT_VERSION, + "schema_version": snapshot["schema_version"], + "contract_version": contract_version_for_schema(str(snapshot["schema_version"])), "cadence": "daily", "as_of": snapshot["as_of"], "generated_at": snapshot["generated_at"], diff --git a/tests/test_advisory_report.py b/tests/test_advisory_report.py index 4a5c2ce..7ca7ab8 100644 --- a/tests/test_advisory_report.py +++ b/tests/test_advisory_report.py @@ -149,9 +149,9 @@ def test_long_horizon_window_is_measured_in_years() -> None: by_symbol = {item["symbol"]: item for item in report["recommendations"]} - assert by_symbol["IDX1"]["primary_horizon"] == "long" - assert by_symbol["IDX1"]["primary_horizon_window"] == "1-3年" - assert "超过3年" in by_symbol["IDX1"]["horizon_note"] + assert by_symbol["IDX1"]["ai_context"]["horizon"] == "long" + assert by_symbol["IDX1"]["ai_context"]["horizon_window"] == "1-3年" + assert by_symbol["IDX1"]["primary_horizon"] == "not_applicable" def test_contract_rejects_execution_enabled_report() -> None: @@ -273,8 +273,8 @@ def test_report_manifest_records_contract_version_and_hashes(tmp_path: Path) -> manifest = json.loads(manifest_path.read_text(encoding="utf-8")) assert manifest["manifest_type"] == "model_recommendation_report" - assert manifest["contract_version"] == "model_recommendations.v5" - assert manifest["version"] == "2026-05-30-weekly-schema-5-run-123-attempt-2" + assert manifest["contract_version"] == "model_recommendations.v6" + assert manifest["version"] == "2026-05-30-weekly-schema-6-run-123-attempt-2" assert manifest["producer"]["git_sha"] == "abcdef1234567890" assert manifest["artifacts"]["json"]["sha256"] assert manifest["artifacts"]["markdown"]["sha256"] diff --git a/tests/test_ai_ingestion_v6.py b/tests/test_ai_ingestion_v6.py new file mode 100644 index 0000000..ef637c8 --- /dev/null +++ b/tests/test_ai_ingestion_v6.py @@ -0,0 +1,311 @@ +from __future__ import annotations + +import datetime as dt +import hashlib +import json +from copy import deepcopy +from pathlib import Path + +import pytest + +from quant_advisor_research import advisory_report as advisory_report_module +from quant_advisor_research.advisory_report import build_advisory_report +from quant_advisor_research.contracts import validate_advisory_report + + +def write_base_inputs(tmp_path: Path) -> tuple[Path, Path]: + tmp_path.mkdir(parents=True, exist_ok=True) + events = tmp_path / "events.csv" + events.write_text( + "event_id,event_date,symbol,event_type,direction,confidence,source_url,notes\n", + encoding="utf-8", + ) + watchlist = tmp_path / "watchlist.csv" + watchlist.write_text( + "symbol,name,bucket,research_status,thesis,source_url\n" + "BASE,Base Candidate,macro_index,watchlist,Base research context,https://example.invalid/base\n", + encoding="utf-8", + ) + return events, watchlist + + +def valid_v2_signal(*, confidence: float = 0.6) -> dict[str, object]: + return { + "schema_version": "2", + "model_version": "shadow-v2", + "scoring_version": "rules-v2", + "as_of": "2026-05-30", + "generated_at": "2026-05-30T12:00:00Z", + "mode": "shadow", + "horizon": "1-3 years", + "universe": ["AI1"], + "regime": "neutral", + "risk_flags": [], + "candidate_bias": { + "AI1": { + "bias": "positive", + "confidence": confidence, + "rationale": "Synthetic advisory-only context.", + } + }, + "confidence": confidence, + "evidence": { + "sources": ["synthetic-source"], + "summary": "Synthetic advisory-only evidence.", + "data_gaps": [], + }, + "expires_at": "2026-06-30", + "policy": { + "execution_allowed": False, + "downstream_use": "Research-only shadow context.", + }, + } + + +def build_report(tmp_path: Path, ai_signal: Path | None = None) -> dict[str, object]: + events, watchlist = write_base_inputs(tmp_path) + return build_advisory_report( + as_of="2026-05-30", + cadence="weekly", + political_events_path=events, + political_watchlist_path=watchlist, + ai_signal_path=ai_signal, + ) + + +def write_signal(path: Path, payload: dict[str, object]) -> Path: + path.write_text(json.dumps(payload) + "\n", encoding="utf-8") + return path + + +def replace_nested(payload: dict[str, object], path: tuple[str, ...], value: object) -> None: + target = payload + for key in path[:-1]: + target = target[key] # type: ignore[assignment] + target[path[-1]] = value + + +def test_v2_signal_requires_versioned_model_metadata_and_fails_closed(tmp_path: Path) -> None: + payload = valid_v2_signal() + payload.pop("model_version") + signal = write_signal(tmp_path / "signal.json", payload) + + report = build_report(tmp_path, signal) + + assert "AI1" not in {item["symbol"] for item in report["recommendations"]} + assert "ai_signal_contract_invalid" in report["summary"]["data_quality_warnings"] + assert report["source_artifacts"]["ai_signal"] == "" + assert report["freshness"]["ai_signal"] == { + "present": False, + "valid": False, + "reason": "not_provided", + } + + +@pytest.mark.parametrize( + ("path", "value"), + [ + (("schema_version",), "3"), + (("generated_at",), "2026-05-30T12:00:00"), + (("mode",), "live"), + (("horizon",), "1-3 months"), + (("candidate_bias",), []), + (("confidence",), True), + (("evidence", "sources"), "not-a-list"), + (("policy", "execution_allowed"), True), + ], +) +def test_ai_contract_type_and_policy_violations_are_no_op( + tmp_path: Path, + path: tuple[str, ...], + value: object, +) -> None: + payload = deepcopy(valid_v2_signal()) + replace_nested(payload, path, value) + signal = write_signal(tmp_path / "signal.json", payload) + + report = build_report(tmp_path, signal) + + assert "AI1" not in {item["symbol"] for item in report["recommendations"]} + assert "ai_signal_contract_invalid" in report["summary"]["data_quality_warnings"] + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("orders", [{"symbol": "AI1"}]), + ("target_weight", 0.5), + ("portfolio_allocation_allowed", True), + ], +) +def test_ai_authority_fields_are_rejected_as_no_op(tmp_path: Path, field: str, value: object) -> None: + payload = valid_v2_signal() + payload[field] = value + signal = write_signal(tmp_path / "signal.json", payload) + + report = build_report(tmp_path, signal) + + assert "AI1" not in {item["symbol"] for item in report["recommendations"]} + assert "ai_signal_contract_invalid" in report["summary"]["data_quality_warnings"] + + +def test_ai_live_downstream_policy_is_rejected_as_no_op(tmp_path: Path) -> None: + payload = valid_v2_signal() + payload["policy"]["downstream_use"] = "live portfolio allocation" + signal = write_signal(tmp_path / "signal.json", payload) + + report = build_report(tmp_path, signal) + + assert "AI1" not in {item["symbol"] for item in report["recommendations"]} + assert "ai_signal_contract_invalid" in report["summary"]["data_quality_warnings"] + + +def test_malformed_ai_json_is_sanitized_no_op(tmp_path: Path) -> None: + signal = tmp_path / "signal.json" + signal.write_text('{"secret": "must-not-leak",', encoding="utf-8") + + report = build_report(tmp_path, signal) + + assert "ai_signal_invalid_json" in report["summary"]["data_quality_warnings"] + assert "must-not-leak" not in json.dumps(report) + assert report["summary"]["ai_regime"] == "not_available" + + +def test_unavailable_ai_signal_is_sanitized_no_op(tmp_path: Path) -> None: + missing_signal = tmp_path / "missing-signal.json" + + report = build_report(tmp_path, missing_signal) + + assert "ai_signal_unavailable" in report["summary"]["data_quality_warnings"] + assert report["source_artifacts"]["ai_signal"] == "" + assert report["freshness"]["ai_signal"] == { + "present": False, + "valid": False, + "reason": "not_provided", + } + assert missing_signal.name not in json.dumps(report) + + +@pytest.mark.parametrize( + ("updates", "reason"), + [ + ({"as_of": "2026-05-20", "generated_at": "2026-05-20T12:00:00Z"}, "stale_as_of"), + ({"as_of": "2026-05-31"}, "as_of_in_future"), + ({"generated_at": "2026-05-31T00:00:00.000001Z"}, "generated_after_reference"), + ( + { + "as_of": "2026-05-20", + "generated_at": "2026-05-20T12:00:00Z", + "expires_at": "2026-05-29", + }, + "expired", + ), + ], +) +def test_temporally_invalid_ai_signal_is_reported_and_does_not_score( + tmp_path: Path, + updates: dict[str, object], + reason: str, +) -> None: + payload = valid_v2_signal() + payload.update(updates) + signal = write_signal(tmp_path / "signal.json", payload) + + report = build_report(tmp_path, signal) + + assert "AI1" not in {item["symbol"] for item in report["recommendations"]} + assert report["freshness"]["ai_signal"]["valid"] is False + assert report["freshness"]["ai_signal"]["reason"] == reason + assert f"ai_signal_{reason}" in report["summary"]["data_quality_warnings"] + + +def test_new_builder_emits_v6_time_contract_and_content_bound_digest(tmp_path: Path) -> None: + first = build_report(tmp_path / "first") + second = build_report(tmp_path / "second") + + generated_at = dt.datetime.fromisoformat(first["generated_at"].replace("Z", "+00:00")) + expires_at = dt.datetime.fromisoformat(first["expires_at"].replace("Z", "+00:00")) + first_events = tmp_path / "first" / "events.csv" + first_watchlist = tmp_path / "first" / "watchlist.csv" + input_identities = { + "political_events": hashlib.sha256(first_events.read_bytes()).hexdigest(), + "political_watchlist": hashlib.sha256(first_watchlist.read_bytes()).hexdigest(), + "ai_signal": None, + "theme_momentum": None, + "market_confirmation": None, + } + expected_digest = hashlib.sha256( + json.dumps(input_identities, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + + assert first["schema_version"] == "6" + assert first["contract_version"] == "model_recommendations.v6" + assert first["reference_time"] == "2026-05-31T00:00:00Z" + assert expires_at == generated_at + dt.timedelta(days=7) + assert first["freshness"] == { + "ai_signal": {"present": False, "valid": False, "reason": "not_provided"}, + "theme_momentum": {"present": False, "valid": False, "reason": "not_provided"}, + } + assert first["input_digest"] == expected_digest + assert second["input_digest"] == first["input_digest"] + validate_advisory_report(first) + + first_watchlist.write_text(first_watchlist.read_text(encoding="utf-8") + "\n", encoding="utf-8") + changed = build_advisory_report( + as_of="2026-05-30", + cadence="weekly", + political_events_path=first_events, + political_watchlist_path=first_watchlist, + ) + assert changed["input_digest"] != first["input_digest"] + + +def test_input_digest_binds_the_bytes_consumed_before_source_replacement(tmp_path: Path, monkeypatch) -> None: + events, watchlist = write_base_inputs(tmp_path) + signal = write_signal(tmp_path / "signal.json", valid_v2_signal()) + original_signal_bytes = signal.read_bytes() + original_loader = advisory_report_module.load_ai_signal + + def replacing_loader(path, *args, **kwargs): + payload = original_loader(path, *args, **kwargs) + replacement = valid_v2_signal() + replacement["candidate_bias"]["AI1"]["bias"] = "negative" + signal.write_text(json.dumps(replacement) + "\n", encoding="utf-8") + return payload + + monkeypatch.setattr(advisory_report_module, "load_ai_signal", replacing_loader) + report = build_advisory_report( + as_of="2026-05-30", + cadence="weekly", + political_events_path=events, + political_watchlist_path=watchlist, + ai_signal_path=signal, + ) + identities = { + "political_events": hashlib.sha256(events.read_bytes()).hexdigest(), + "political_watchlist": hashlib.sha256(watchlist.read_bytes()).hexdigest(), + "ai_signal": hashlib.sha256(original_signal_bytes).hexdigest(), + "theme_momentum": None, + "market_confirmation": None, + } + expected_digest = hashlib.sha256( + json.dumps(identities, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + + assert report["input_digest"] == expected_digest + assert next(item for item in report["recommendations"] if item["symbol"] == "AI1")["ai_context"]["bias"] == "positive" + + +def test_positive_ai_confidence_is_display_only_for_recommendation_scoring(tmp_path: Path) -> None: + low_path = write_signal(tmp_path / "low.json", valid_v2_signal(confidence=0.0)) + high_path = write_signal(tmp_path / "high.json", valid_v2_signal(confidence=1.0)) + low = build_report(tmp_path / "low", low_path) + high = build_report(tmp_path / "high", high_path) + + low_rec = next(item for item in low["recommendations"] if item["symbol"] == "AI1") + high_rec = next(item for item in high["recommendations"] if item["symbol"] == "AI1") + + assert low_rec["ai_context"]["confidence"] == 0.0 + assert high_rec["ai_context"]["confidence"] == 1.0 + for field in ("evidence_score", "risk_score", "rating", "score", "long_horizon_ai_score"): + assert low_rec[field] == high_rec[field] diff --git a/tests/test_artifact_integrity.py b/tests/test_artifact_integrity.py index 6ba952c..77a42c6 100644 --- a/tests/test_artifact_integrity.py +++ b/tests/test_artifact_integrity.py @@ -35,7 +35,11 @@ def build_report(*, schema_version: str = "5") -> dict: political_events_path=ROOT / "examples/political_events.example.csv", political_watchlist_path=ROOT / "examples/political_watchlist.example.csv", ) - if schema_version == "6": + if schema_version == "5": + report.update(schema_version="5") + for key in ("contract_version", "reference_time", "expires_at", "freshness", "input_digest"): + report.pop(key) + else: generated_at = normalize_aware_datetime(report["generated_at"]) reference_time = canonical_reference_time(dt.date.fromisoformat(report["as_of"])) report.update( diff --git a/tests/test_build_pipeline.py b/tests/test_build_pipeline.py index 668995c..43fe8e1 100644 --- a/tests/test_build_pipeline.py +++ b/tests/test_build_pipeline.py @@ -8,7 +8,12 @@ from quant_advisor_research.archive_backfill import backfill_site_archive, discover_report_paths from quant_advisor_research import build_pipeline as build_pipeline_module -from quant_advisor_research.build_pipeline import build_advisory_artifacts, default_weekly_as_of +from quant_advisor_research.build_pipeline import ( + build_advisory_artifacts, + default_daily_as_of, + default_monthly_as_of, + default_weekly_as_of, +) from quant_advisor_research.cross_repo_smoke import run_cross_repo_smoke @@ -43,10 +48,13 @@ def build_fixture_report(tmp_path: Path, as_of: dt.date) -> Path: return result.report_json -def test_default_weekly_as_of_uses_most_recent_saturday() -> None: - assert default_weekly_as_of(dt.date(2026, 6, 20)) == dt.date(2026, 6, 20) +def test_default_as_of_values_use_only_closed_periods() -> None: + assert default_daily_as_of(dt.date(2026, 6, 20)) == dt.date(2026, 6, 19) + assert default_weekly_as_of(dt.date(2026, 6, 20)) == dt.date(2026, 6, 13) assert default_weekly_as_of(dt.date(2026, 6, 21)) == dt.date(2026, 6, 20) assert default_weekly_as_of(dt.date(2026, 6, 24)) == dt.date(2026, 6, 20) + assert default_monthly_as_of(dt.date(2026, 6, 1)) == dt.date(2026, 5, 31) + assert default_monthly_as_of(dt.date(2026, 6, 20)) == dt.date(2026, 5, 31) def test_build_advisory_artifacts_builds_market_report_and_site(tmp_path: Path) -> None: diff --git a/tests/test_d3_exact_bundle.py b/tests/test_d3_exact_bundle.py index 6c08cf5..6c421b5 100644 --- a/tests/test_d3_exact_bundle.py +++ b/tests/test_d3_exact_bundle.py @@ -6,7 +6,7 @@ import pytest sys.path.insert(0, str(Path(__file__).parents[1] / "scripts")) -from d3_evidence import EvidenceContractError, canonical_distribution_snapshot, validate_exact_bundle +from d3_evidence import EvidenceContractError, canonical_distribution_snapshot, preview_source_contract, validate_exact_bundle def _normal_bundle(path: Path) -> None: @@ -68,6 +68,25 @@ def test_distribution_snapshot_uses_pep503_and_rejects_conflict(): canonical_distribution_snapshot(["alpha_.==1.0", "ALPHA-==2.0"]) +def test_preview_source_contract_accepts_v5_and_v6() -> None: + for schema_version, contract_version in ( + ("5", "model_recommendations.v5"), + ("6", "model_recommendations.v6"), + ): + source = { + "schema_version": schema_version, + "contract_version": contract_version, + "cadence": "daily", + "as_of": "2026-05-30", + "generated_at": "2026-05-31T00:00:00Z", + } + assert preview_source_contract( + {"source": source}, + as_of="2026-05-30", + generated_at="2026-05-31T00:00:00Z", + ) == source + + def test_workflow_uses_exact_paths_and_immutable_actions(): workflow = (Path(__file__).parents[1] / ".github/workflows/qar_d3_daily_preview_artifact.yml").read_text() assert "pull_request:" in workflow and "workflow_dispatch:" in workflow diff --git a/tests/test_m0_research_hypothesis.py b/tests/test_m0_research_hypothesis.py index c910a6d..af0359c 100644 --- a/tests/test_m0_research_hypothesis.py +++ b/tests/test_m0_research_hypothesis.py @@ -52,8 +52,8 @@ def test_adapter_projects_public_report_to_closed_research_only_hypotheses() -> dict.fromkeys(report["recommendations"][0]["suitable_horizons"]) ) assert first["research_context"]["primary_horizon"] in first["research_context"]["suitable_horizons"] - assert first["provenance"]["source_schema_version"] == "5" - assert first["provenance"]["source_input_digest"] is None + assert first["provenance"]["source_schema_version"] == "6" + assert first["provenance"]["source_input_digest"] == report["input_digest"] assert first["expires_at"] > first["generated_at"] assert "target_weight" not in first assert "recommendations" not in first diff --git a/tests/test_preview_bundle.py b/tests/test_preview_bundle.py index 044be71..7d2497e 100644 --- a/tests/test_preview_bundle.py +++ b/tests/test_preview_bundle.py @@ -30,7 +30,8 @@ def report(*, cadence="daily", as_of="2026-06-20"): def test_daily_bundle_builds_and_readback_validates(tmp_path): output = tmp_path / "preview" - result = build_preview_bundle(report(), output) + value = report() + result = build_preview_bundle(value, output) assert result.bundle_contract == BUNDLE_CONTRACT assert sorted(path.name for path in output.iterdir()) == ["manifest.json", "report.html", "report.json"] @@ -40,11 +41,11 @@ def test_daily_bundle_builds_and_readback_validates(tmp_path): assert manifest == { "bundle_contract": BUNDLE_CONTRACT, "source": { - "schema_version": "5", + "schema_version": "6", "contract_version": SOURCE_CONTRACT_VERSION, "cadence": "daily", "as_of": "2026-06-20", - "generated_at": report()["generated_at"], + "generated_at": value["generated_at"], }, "artifacts": { "report.json": { @@ -70,7 +71,7 @@ def test_non_daily_source_is_rejected_before_output(cadence, tmp_path): @pytest.mark.parametrize("mutation", [ - lambda value: value.update(schema_version="6"), + lambda value: value.update(schema_version="7"), lambda value: value.update(contract_version="wrong"), lambda value: value.update(generated_at="not-a-datetime"), ]) @@ -83,6 +84,17 @@ def test_source_contract_mutations_fail_closed_without_partial_output(mutation, assert not output.exists() +def test_legacy_v5_source_remains_readable(tmp_path): + value = report() + value.update(schema_version="5") + for key in ("contract_version", "reference_time", "expires_at", "freshness", "input_digest"): + value.pop(key) + + build_preview_bundle(value, tmp_path / "preview") + + assert read_preview_bundle(tmp_path / "preview").report["schema_version"] == "5" + + def test_build_is_deterministic_for_equivalent_mapping_order(tmp_path): left = tmp_path / "left" right = tmp_path / "right" diff --git a/tests/test_preview_workspace.py b/tests/test_preview_workspace.py index 29d1b3e..130c2eb 100644 --- a/tests/test_preview_workspace.py +++ b/tests/test_preview_workspace.py @@ -2,6 +2,7 @@ import inspect import os +from copy import deepcopy from pathlib import Path import pytest @@ -38,8 +39,9 @@ def test_workspace_is_private_exact_three_files_and_readback_valid(tmp_path): def test_equivalent_reports_produce_identical_bytes_in_distinct_workspaces(tmp_path): - first = build_preview_workspace(report(), tmp_path) - second = build_preview_workspace(report(), tmp_path) + payload = report() + first = build_preview_workspace(payload, tmp_path) + second = build_preview_workspace(deepcopy(payload), tmp_path) assert first != second assert {p.name: p.read_bytes() for p in first.iterdir()} == { diff --git a/tests/test_publication_plan.py b/tests/test_publication_plan.py index eded3d1..c137f17 100644 --- a/tests/test_publication_plan.py +++ b/tests/test_publication_plan.py @@ -1,6 +1,7 @@ from __future__ import annotations import dataclasses +import datetime as dt import hashlib from pathlib import Path @@ -21,6 +22,7 @@ build_publication_plan, ) from quant_advisor_research.time_contract import contract_version_for_schema +from quant_advisor_research.time_contract import normalize_aware_datetime ROOT = Path(__file__).resolve().parents[1] @@ -36,6 +38,8 @@ def report(*, as_of: str = "2026-06-20", generated_at: str | None = None) -> dic ) if generated_at is not None: value["generated_at"] = generated_at + generated = normalize_aware_datetime(generated_at) + value["expires_at"] = (generated + dt.timedelta(days=7)).isoformat().replace("+00:00", "Z") return value diff --git a/tests/test_publisher.py b/tests/test_publisher.py index 20015a6..d529322 100644 --- a/tests/test_publisher.py +++ b/tests/test_publisher.py @@ -131,6 +131,9 @@ def test_publish_reports_writes_site_files(tmp_path: Path) -> None: def make_report_for_date(base_report: dict, as_of: str) -> dict: report = deepcopy(base_report) + report.update(schema_version="5", contract_version="model_recommendations.v5") + for key in ("reference_time", "expires_at", "freshness", "input_digest"): + report.pop(key) report["as_of"] = as_of report["generated_at"] = f"{as_of}T00:00:00+00:00" return report diff --git a/tests/test_publisher_period_redesign.py b/tests/test_publisher_period_redesign.py index 0aaac3d..de11ca3 100644 --- a/tests/test_publisher_period_redesign.py +++ b/tests/test_publisher_period_redesign.py @@ -25,12 +25,16 @@ def build_v5(as_of: str = "2026-06-20") -> dict: - return build_advisory_report( + report = build_advisory_report( as_of=as_of, cadence="weekly", political_events_path=ROOT / "examples/political_events.example.csv", political_watchlist_path=ROOT / "examples/political_watchlist.example.csv", ) + report.update(schema_version="5", contract_version="model_recommendations.v5") + for key in ("reference_time", "expires_at", "freshness", "input_digest"): + report.pop(key) + return report def build_v6(as_of: str = "2026-06-20") -> dict: diff --git a/tests/test_publisher_redesign_stage2a.py b/tests/test_publisher_redesign_stage2a.py index 9218bc6..0e4a0c7 100644 --- a/tests/test_publisher_redesign_stage2a.py +++ b/tests/test_publisher_redesign_stage2a.py @@ -19,12 +19,19 @@ def build_v5() -> dict: - return build_advisory_report( + report = build_advisory_report( as_of="2026-05-30", cadence="weekly", political_events_path=ROOT / "examples/political_events.example.csv", political_watchlist_path=ROOT / "examples/political_watchlist.example.csv", ) + report.update( + schema_version="5", + contract_version="model_recommendations.v5", + ) + for key in ("reference_time", "expires_at", "freshness", "input_digest"): + report.pop(key) + return report def build_v6() -> dict: diff --git a/tests/test_workflow_defaults.py b/tests/test_workflow_defaults.py index 6d1359d..9d219c4 100644 --- a/tests/test_workflow_defaults.py +++ b/tests/test_workflow_defaults.py @@ -43,3 +43,16 @@ def test_cross_repo_smoke_workflow_uses_live_artifacts_and_no_network_market_fal assert "data/output/latest_signal.json" in text assert "data/output/theme_momentum_snapshot.json" in text assert "scripts/run_cross_repo_smoke.py" in text + + +def test_workflow_default_as_of_uses_only_closed_period_helpers() -> None: + expected_helpers = { + "weekly_advisory_review.yml": "default_weekly_as_of", + "publish_advisory_site.yml": "default_weekly_as_of", + "monthly_advisory_review.yml": "default_monthly_as_of", + "cross_repo_smoke.yml": "default_daily_as_of", + } + for workflow, helper in expected_helpers.items(): + text = (ROOT / ".github" / "workflows" / workflow).read_text(encoding="utf-8") + assert helper in text + assert "date -u +%F" not in text