# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca """Typed contracts shared by the data layer, the agents, the debate protocols and the evaluation. Conventions ----------- * Money in CAD, areas in m2, distances in km, dates ISO (YYYY-MM-DD). * Every piece of evidence shown to an agent carries an ``evidence_id`` (``P`` = subject property, ``C01..`` = comparables, ``M*`` = market context, ``G*`` = geospatial, ``R*`` = roll/assessment). Agents must cite these ids; the evaluator checks that cited ids exist in the packet. * The subject's transaction price is NEVER part of an evidence packet. """ from __future__ import annotations import hashlib import json from datetime import date from typing import Any, Literal from pydantic import BaseModel, Field, field_validator, model_validator PropertyClass = Literal["house", "condo", "plex"] Subtype = Literal["detached", "semi", "row", "integrated", "unknown"] class SubjectProperty(BaseModel): """What is known about the subject *as of the valuation date* (no price).""" evidence_id: str = "P" tx_id: str valuation_date: date information_cutoff: date = Field(description="Agents may only use evidence dated on or before this date") property_class: PropertyClass subtype: Subtype = "unknown" region: str municipality: str borough: str | None = None lat: float = Field(description="rounded to 3 decimals (~100 m) for privacy") lng: float floor_area_m2: float | None = None land_area_m2: float | None = None year_built: int | None = None storeys: float | None = None n_units: int | None = None roll_land_value: float | None = None roll_building_value: float | None = None roll_total_value: float | None = None roll_reference_date: str | None = Field(default=None, description="market reference date of the roll in force (e.g. 2022-07-01)") prior_sale_price: float | None = Field(default=None, description="most recent *earlier* sale of the same parcel, if any (strictly before cutoff)") prior_sale_date: date | None = None new_construction: bool | None = None missing_fields: list[str] = Field(default_factory=list) class Comparable(BaseModel): evidence_id: str = Field(pattern=r"^C\d{2}$") sale_date: date price: float distance_km: float property_class: PropertyClass subtype: Subtype = "unknown" municipality: str floor_area_m2: float | None = None land_area_m2: float | None = None year_built: int | None = None storeys: float | None = None n_units: int | None = None roll_total_value: float | None = None price_per_m2: float | None = None assessment_ratio: float | None = Field(default=None, description="price / roll_total_value") months_before_valuation: float class MarketContext(BaseModel): evidence_id: str = "M1" as_of: date municipality_median_price_12m: float | None = None municipality_median_ppm2_12m: float | None = None municipality_n_sales_12m: int | None = None municipality_price_index: list[dict[str, Any]] = Field(default_factory=list, description="[{month, index}] up to as_of (base = first month of sample)") region_median_price_12m: float | None = None region_yoy_change: float | None = None local_liquidity_n_sales_3km_12m: int | None = None local_median_assessment_ratio_12m: float | None = Field(default=None, description="median price/roll ratio within 3 km over the last 12 months") mortgage_rate_5y_fixed: float | None = Field(default=None, description="not available in this snapshot unless loaded from an external series") class GeoContext(BaseModel): evidence_id: str = "G1" in_flood_zone: bool | None = None transit_stops_within_1km: int | None = None da_population_density: float | None = None da_median_income: float | None = None distance_to_cbd_km: float | None = None notes: list[str] = Field(default_factory=list) class EvidencePacket(BaseModel): packet_id: str subject: SubjectProperty comparables: list[Comparable] market: MarketContext geo: GeoContext | None = None role_visible_sections: list[str] = Field(default_factory=lambda: ["subject", "comparables", "market", "geo"]) notes: list[str] = Field(default_factory=list) def evidence_ids(self) -> set[str]: ids = {self.subject.evidence_id, self.market.evidence_id} ids |= {c.evidence_id for c in self.comparables} if self.geo is not None: ids.add(self.geo.evidence_id) return ids def content_hash(self) -> str: return hashlib.sha256(self.model_dump_json(exclude={"packet_id"}).encode()).hexdigest()[:16] class Adjustment(BaseModel): comparable_id: str adjusted_price: float = Field(description="comparable price after all adjustments, CAD") reason: str class AgentEstimate(BaseModel): """Structured output every agent must return (the JSON contract).""" point_estimate: float = Field(description="most likely market value in CAD as of the valuation date") q05: float q25: float q50: float q75: float q95: float evidence_ids_used: list[str] = Field(description="ids from the packet actually relied upon") adjustments: list[Adjustment] = Field(default_factory=list) key_assumptions: list[str] = Field(default_factory=list) missing_data: list[str] = Field(default_factory=list) confidence_rationale: str = "" counterarguments: list[str] = Field(default_factory=list) reasoning_summary: str = Field(default="", description="<= 120 words") @model_validator(mode="after") def _check_quantiles(self) -> "AgentEstimate": qs = [self.q05, self.q25, self.q50, self.q75, self.q95] if any(q <= 0 for q in qs) or self.point_estimate <= 0: raise ValueError("estimates must be positive") if any(b < a for a, b in zip(qs, qs[1:])): raise ValueError("quantiles must be non-decreasing") if not (self.q05 <= self.point_estimate <= self.q95): raise ValueError("point estimate must lie within [q05, q95]") return self class Critique(BaseModel): target_role: str agree: bool points: list[str] evidence_ids_cited: list[str] = Field(default_factory=list) suggested_value: float | None = None class CritiqueSet(BaseModel): critiques: list[Critique] class AgentTurn(BaseModel): role: str round: int kind: Literal["estimate", "critique", "revision"] model: str provider: str estimate: AgentEstimate | None = None critiques: CritiqueSet | None = None input_tokens: int = 0 output_tokens: int = 0 cached_tokens: int = 0 cost_usd: float = 0.0 latency_s: float = 0.0 prompt_hash: str = "" request_id: str | None = None validation_retries: int = 0 error: str | None = None class DebateTranscript(BaseModel): tx_id: str arm: str committee: str protocol: str rounds: int packet_hash: str turns: list[AgentTurn] final_estimates: dict[str, float] = Field(default_factory=dict, description="role -> final point estimate") final_quantiles: dict[str, list[float]] = Field(default_factory=dict, description="role -> [q05,q25,q50,q75,q95]") aggregated: dict[str, float] = Field(default_factory=dict, description="aggregation rule -> value") aggregated_interval: dict[str, list[float]] = Field(default_factory=dict, description="rule -> [lo90, hi90]") total_cost_usd: float = 0.0 total_latency_s: float = 0.0 total_tokens: int = 0 assignment: dict[str, str] = Field(default_factory=dict, description="role -> model id (committee assignment, recorded per property)") notes: list[str] = Field(default_factory=list, description="protocol events: stopped_early_at=, verification flags, blinded skeptic, ...") def to_json(self) -> str: return self.model_dump_json(indent=1) def canonical_json(obj: Any) -> str: return json.dumps(obj, sort_keys=True, ensure_ascii=False, default=str)