BuyingAgent/backend/app/scoring/engine.py

546 lines
20 KiB
Python
Raw Permalink Normal View History

from collections.abc import Sequence
from datetime import datetime, timedelta
from decimal import Decimal
from typing import Any
from app.domain.attributes import select_attribute_snapshot
from app.domain.constraints import evaluate_constraint
from app.domain.enums import (
Availability,
ConstraintStatus,
Eligibility,
PriceReferenceScope,
)
from app.domain.models import (
AttributeDefinition,
AttributeStatement,
ConstraintResult,
EvaluationResult,
GateResult,
Offer,
PriceObservation,
Product,
ProductVariant,
Profile,
Reason,
ScoreComponent,
ScoringModel,
)
from app.domain.serialize import jsonable
from app.scoring.deal import classification_for, derive_eligible, eligibility_from_parts, score_deal
from app.scoring.mathutil import round_score
from app.scoring.price import (
build_offer_reference_history,
price_statistics_from_windows,
score_price,
select_current_observation,
)
from app.scoring.product import score_product
def evaluate_deal(
*,
profile: Profile,
scoring_model: ScoringModel,
product: Product,
variant: ProductVariant,
offer: Offer,
attributes: tuple[AttributeStatement, ...],
observations: tuple[PriceObservation, ...],
definitions: dict[str, AttributeDefinition],
as_of: datetime,
snapshot_cutoff: datetime,
engine_build: str,
) -> EvaluationResult:
observations = tuple(item for item in observations if item.offer_id == offer.id)
snapshot = select_attribute_snapshot(attributes, as_of=as_of, snapshot_cutoff=snapshot_cutoff)
constraint_results = tuple(
evaluate_constraint(
constraint,
snapshot.get(constraint.attribute),
definitions[constraint.attribute],
scoring_model.minimum_confidence,
)
for constraint in profile.constraints
)
product_gates, product_eligibility = _product_eligibility(profile, product, constraint_results)
product_score_raw, coverage, preference_components = score_product(
profile.preferences,
snapshot,
constraint_results,
scoring_model,
)
current = select_current_observation(observations, as_of=as_of, snapshot_cutoff=snapshot_cutoff)
offer_gates, offer_eligibility = _offer_eligibility(profile, offer, current, scoring_model, as_of)
overall = eligibility_from_parts(product_eligibility, offer_eligibility)
windows: dict[int, Any] = {}
price_score_raw: Decimal | None = None
budget_score: Decimal | None = None
history_score: Decimal | None = None
price_basis = None
history_warning: str | None = None
if _can_score_price(current, profile, scoring_model, as_of):
windows = build_offer_reference_history(observations, current, scoring_model)
price_score_raw, budget_score, history_score, price_basis, history_warning = score_price(
current_price=current.total_price,
reference_history=windows[90].values,
budget=profile.budget,
model=scoring_model,
)
stats = price_statistics_from_windows(current, windows)
deal_score_raw = None
if overall == Eligibility.ELIGIBLE and price_score_raw is not None:
deal_score_raw = score_deal(product_score_raw, price_score_raw, scoring_model)
classification = classification_for(overall, deal_score_raw, scoring_model)
reasons, warnings = _reasons(
constraint_results,
product_gates,
offer_gates,
stats,
current,
history_warning,
price_score_raw,
)
components = _all_components(
preference_components,
product_gates,
offer_gates,
budget_score,
history_score,
scoring_model,
)
input_snapshot = {
"profile": profile.definition,
"scoring_model": scoring_model.definition,
"product": jsonable(_product_dict(product)),
"variant": jsonable(_variant_dict(variant)),
"offer": jsonable(_offer_dict(offer)),
"attribute_snapshot": {
key: jsonable(_attribute_dict(value))
for key, value in snapshot.items()
},
"current_observation": None if current is None else jsonable(_observation_dict(current)),
"history_observation_ids": [str(item) for item in stats.used_observation_ids],
"as_of": jsonable(as_of),
"snapshot_cutoff": jsonable(snapshot_cutoff),
"price_reference_scope": PriceReferenceScope.OFFER.value,
"unrounded": {
"product_score": jsonable(product_score_raw),
"price_score": jsonable(price_score_raw),
"deal_score": jsonable(deal_score_raw),
"budget_score": jsonable(budget_score),
"history_score": jsonable(history_score),
},
}
return EvaluationResult(
profile_id=profile.id,
profile_version=profile.version,
scoring_version=scoring_model.version,
scoring_hash=scoring_model.definition_hash,
engine_build=engine_build,
variant_id=variant.id,
offer_id=offer.id,
as_of=as_of,
snapshot_cutoff=snapshot_cutoff,
product_eligibility=product_eligibility,
offer_eligibility=offer_eligibility,
overall_eligibility=overall,
eligible=derive_eligible(overall),
product_score=round_score(product_score_raw),
product_score_raw=product_score_raw,
price_score=None if price_score_raw is None else round_score(price_score_raw),
price_score_raw=price_score_raw,
deal_score=None if deal_score_raw is None else round_score(deal_score_raw),
deal_score_raw=deal_score_raw,
classification=classification,
price_basis=price_basis,
product_coverage=coverage,
constraint_results=constraint_results,
product_gates=product_gates,
offer_gates=offer_gates,
components=components,
price_statistics=stats,
price_reference_scope=PriceReferenceScope.OFFER,
reasons=reasons,
warnings=warnings,
input_snapshot=input_snapshot,
failed_constraints=tuple(
item for item in constraint_results if item.status == ConstraintStatus.FAIL.value
),
unknown_constraints=tuple(
item for item in constraint_results if item.status == ConstraintStatus.UNKNOWN.value
),
)
def statement_price_confidence(observation: PriceObservation, model: ScoringModel) -> bool:
return observation.confidence >= model.minimum_confidence
def _fresh_enough(observation: PriceObservation, as_of: datetime, model: ScoringModel) -> bool:
return as_of - observation.observed_at <= timedelta(hours=model.maximum_price_age_hours)
def _product_eligibility(
profile: Profile,
product: Product,
constraint_results: tuple[ConstraintResult, ...],
) -> tuple[tuple[GateResult, ...], Eligibility]:
category_ok = product.category == profile.category
category_gate = GateResult(
key="category",
status="PASS" if category_ok else "FAIL",
reason_code="CATEGORY_MATCH" if category_ok else "CATEGORY_MISMATCH",
details={"expected": profile.category, "actual": product.category},
)
mandatory = [item for item in constraint_results if item.mandatory]
if not category_ok or any(item.status == ConstraintStatus.FAIL.value for item in mandatory):
eligibility = Eligibility.INELIGIBLE
elif any(item.status == ConstraintStatus.UNKNOWN.value for item in mandatory):
eligibility = Eligibility.UNKNOWN
else:
eligibility = Eligibility.ELIGIBLE
return (category_gate,), eligibility
def _offer_eligibility(
profile: Profile,
offer: Offer,
current: PriceObservation | None,
model: ScoringModel,
as_of: datetime,
) -> tuple[tuple[GateResult, ...], Eligibility]:
gates: list[GateResult] = []
market_ok = offer.market == profile.market
gates.append(_gate("market", market_ok, "MARKET_MATCH", "MARKET_MISMATCH", offer.market, profile.market))
condition_ok = offer.condition in profile.allowed_conditions
gates.append(
_gate(
"condition",
condition_ok,
"CONDITION_ALLOWED",
"CONDITION_NOT_ALLOWED",
offer.condition.value,
[item.value for item in profile.allowed_conditions],
)
)
currency_ok = offer.currency == profile.currency
gates.append(_gate("currency", currency_ok, "CURRENCY_MATCH", "CURRENCY_MISMATCH", offer.currency, profile.currency))
if current is None:
gates.append(GateResult("price_present", "UNKNOWN", "PRICE_MISSING", {}))
gates.append(GateResult("price_fresh", "UNKNOWN", "PRICE_MISSING", {}))
gates.append(GateResult("price_confidence", "UNKNOWN", "PRICE_MISSING", {}))
gates.append(GateResult("shipping", "UNKNOWN", "SHIPPING_UNKNOWN", {}))
gates.append(GateResult("budget", "UNKNOWN", "PRICE_MISSING", {}))
gates.append(GateResult("availability", "UNKNOWN", "AVAILABILITY_UNKNOWN", {}))
return tuple(gates), _combine_offer_gates(gates)
fresh = _fresh_enough(current, as_of, model)
if not fresh:
gates.append(GateResult("price_present", "UNKNOWN", "PRICE_STALE", {"observed_at": current.observed_at.isoformat()}))
gates.append(GateResult("price_fresh", "UNKNOWN", "PRICE_STALE", {}))
else:
gates.append(GateResult("price_present", "PASS", "PRICE_PRESENT", {"observation_id": str(current.id)}))
gates.append(GateResult("price_fresh", "PASS", "PRICE_FRESH", {}))
confident = statement_price_confidence(current, model)
gates.append(
GateResult(
"price_confidence",
"PASS" if confident else "UNKNOWN",
"PRICE_CONFIDENCE_OK" if confident else "PRICE_LOW_CONFIDENCE",
{"confidence": str(current.confidence)},
)
)
shipping_known = current.shipping_price is not None
gates.append(
GateResult(
"shipping",
"PASS" if shipping_known else "UNKNOWN",
"SHIPPING_KNOWN" if shipping_known else "SHIPPING_UNKNOWN",
{},
)
)
total = current.total_price
if total is None or not currency_ok:
gates.append(GateResult("budget", "UNKNOWN", "PRICE_INCOMPLETE", {}))
elif total <= profile.budget.maximum:
gates.append(GateResult("budget", "PASS", "WITHIN_BUDGET", {"total": str(total)}))
else:
gates.append(GateResult("budget", "FAIL", "OVER_BUDGET", {"total": str(total)}))
if current.availability == Availability.IN_STOCK:
gates.append(GateResult("availability", "PASS", "IN_STOCK", {}))
elif current.availability == Availability.OUT_OF_STOCK:
gates.append(GateResult("availability", "FAIL", "OUT_OF_STOCK", {}))
else:
gates.append(GateResult("availability", "UNKNOWN", "AVAILABILITY_UNKNOWN", {}))
if current.currency != offer.currency:
gates.append(GateResult("observation_currency", "FAIL", "CURRENCY_MISMATCH", {}))
if current.source_id != offer.source_id:
gates.append(GateResult("observation_source", "FAIL", "SOURCE_MISMATCH", {}))
return tuple(gates), _combine_offer_gates(gates)
def _gate(key: str, ok: bool, pass_code: str, fail_code: str, actual: Any, expected: Any) -> GateResult:
return GateResult(
key=key,
status="PASS" if ok else "FAIL",
reason_code=pass_code if ok else fail_code,
details={"actual": actual, "expected": expected},
)
def _can_score_price(
current: PriceObservation | None,
profile: Profile,
model: ScoringModel,
as_of: datetime,
) -> bool:
if current is None:
return False
if current.currency != profile.currency:
return False
if current.total_price is None:
return False
if not _fresh_enough(current, as_of, model):
return False
return statement_price_confidence(current, model)
def _combine_offer_gates(gates: Sequence[GateResult]) -> Eligibility:
statuses = [gate.status for gate in gates]
if "FAIL" in statuses:
return Eligibility.INELIGIBLE
if "UNKNOWN" in statuses:
return Eligibility.UNKNOWN
return Eligibility.ELIGIBLE
def _reasons(
constraints: tuple[ConstraintResult, ...],
product_gates: tuple[GateResult, ...],
offer_gates: tuple[GateResult, ...],
stats,
current: PriceObservation | None,
history_warning: str | None,
price_score_raw: Decimal | None,
) -> tuple[tuple[Reason, ...], tuple[str, ...]]:
reasons: list[Reason] = []
warnings: list[str] = []
for gate in (*product_gates, *offer_gates):
if gate.status != "PASS":
reasons.append(Reason(gate.reason_code, _gate_message(gate)))
for item in constraints:
if item.status != ConstraintStatus.PASS.value:
reasons.append(Reason(item.reason_code, _constraint_message(item)))
if history_warning:
warnings.append(history_warning)
reasons.append(Reason(history_warning, "Zu wenige Historientage für den 90-Tage-Median."))
if stats.lowest_365d is not None and not stats.lowest_365d_is_full_year:
warnings.append("LOWEST_365D_PARTIAL")
reasons.append(
Reason(
"LOWEST_365D_PARTIAL",
f"lowest_365d {stats.lowest_365d} ist das Minimum der vorhandenen {stats.lowest_365d_days} Tage, kein belegtes Jahrestief.",
)
)
if (
price_score_raw is not None
and current is not None
and current.total_price is not None
and stats.median_90d is not None
and stats.median_90d != 0
):
delta = (stats.median_90d - current.total_price) / stats.median_90d * Decimal("100")
percent = round_score(delta)
if current.total_price < stats.median_90d:
reasons.append(
Reason(
"PRICE_BELOW_MEDIAN",
f"Gesamtpreis {_de_decimal(percent)} % unter dem Median der beobachteten 90-Tage-Historie.",
)
)
elif current.total_price > stats.median_90d:
reasons.append(
Reason(
"PRICE_ABOVE_MEDIAN",
f"Gesamtpreis {_de_decimal(percent.copy_abs())} % über dem Median der beobachteten 90-Tage-Historie.",
)
)
else:
reasons.append(Reason("PRICE_AT_MEDIAN", "Gesamtpreis entspricht dem Median der 90-Tage-Historie."))
return tuple(reasons), tuple(warnings)
def _de_decimal(value: Decimal) -> str:
return format(value, "f").replace(".", ",")
def _gate_message(gate: GateResult) -> str:
messages = {
"CATEGORY_MISMATCH": "Die Produktkategorie entspricht nicht dem Profil.",
"MARKET_MISMATCH": "Der Angebotsmarkt entspricht nicht dem Profil.",
"CONDITION_NOT_ALLOWED": "Der Angebotszustand ist im Profil nicht erlaubt.",
"CURRENCY_MISMATCH": "Die Währung entspricht nicht dem Profil.",
"PRICE_MISSING": "Es liegt keine aktuelle Preisbeobachtung vor.",
"PRICE_STALE": "Die aktuelle Preisbeobachtung ist nicht frisch genug.",
"PRICE_LOW_CONFIDENCE": "Die Preis-Confidence liegt unter der Mindestschwelle.",
"SHIPPING_UNKNOWN": "Obligatorische Versandkosten sind unbekannt.",
"OVER_BUDGET": "Der Gesamtpreis liegt über dem Budgetmaximum.",
"OUT_OF_STOCK": "Das Angebot ist nicht lieferbar.",
"AVAILABILITY_UNKNOWN": "Die Verfügbarkeit ist unbekannt.",
"SOURCE_MISMATCH": "Die Beobachtungsquelle weicht vom Angebot ab.",
}
return messages.get(gate.reason_code, gate.reason_code)
def _constraint_message(item: ConstraintResult) -> str:
if item.status == ConstraintStatus.FAIL.value:
return f"Constraint {item.constraint_id} ist nicht erfüllt."
return f"Constraint {item.constraint_id} kann nicht bewertet werden."
def _all_components(
preference_components: tuple[ScoreComponent, ...],
product_gates: tuple[GateResult, ...],
offer_gates: tuple[GateResult, ...],
budget_score: Decimal | None,
history_score: Decimal | None,
model: ScoringModel,
) -> tuple[ScoreComponent, ...]:
extra: list[ScoreComponent] = list(preference_components)
if budget_score is not None:
extra.append(
ScoreComponent(
kind="budget",
component_key="budget",
raw_score=budget_score,
weight=model.budget_weight,
contribution=model.budget_weight * budget_score,
details={},
)
)
if history_score is not None:
extra.append(
ScoreComponent(
kind="history",
component_key="history",
raw_score=history_score,
weight=model.history_weight,
contribution=model.history_weight * history_score,
details={},
)
)
for gate in product_gates:
extra.append(
ScoreComponent(
kind="product_gate",
component_key=gate.key,
raw_score=None,
weight=None,
contribution=None,
details={"status": gate.status, "reason_code": gate.reason_code, **gate.details},
)
)
for gate in offer_gates:
extra.append(
ScoreComponent(
kind="offer_gate",
component_key=gate.key,
raw_score=None,
weight=None,
contribution=None,
details={"status": gate.status, "reason_code": gate.reason_code, **gate.details},
)
)
return tuple(extra)
def _product_dict(product: Product) -> dict[str, Any]:
return {
"id": product.id,
"category": product.category,
"brand": product.brand,
"name": product.name,
"model": product.model,
"created_at": product.created_at,
}
def _variant_dict(variant: ProductVariant) -> dict[str, Any]:
return {
"id": variant.id,
"product_id": variant.product_id,
"name": variant.name,
"manufacturer_sku": variant.manufacturer_sku,
"identifiers": variant.identifiers,
"created_at": variant.created_at,
}
def _offer_dict(offer: Offer) -> dict[str, Any]:
return {
"id": offer.id,
"variant_id": offer.variant_id,
"source_id": offer.source_id,
"external_id": offer.external_id,
"seller": offer.seller,
"url": offer.url,
"market": offer.market,
"currency": offer.currency,
"condition": offer.condition,
"created_at": offer.created_at,
}
def _attribute_dict(statement: AttributeStatement) -> dict[str, Any]:
return {
"id": statement.id,
"variant_id": statement.variant_id,
"attribute": statement.attribute,
"raw_value": statement.raw_value,
"normalized_value": statement.normalized_value,
"unit": statement.unit,
"normalization_status": statement.normalization_status,
"normalizer_version": statement.normalizer_version,
"confidence": statement.confidence,
"source_id": statement.source_id,
"source_reference": statement.source_reference,
"observed_at": statement.observed_at,
"ingested_at": statement.ingested_at,
"normalization_error": statement.normalization_error,
}
def _observation_dict(observation: PriceObservation) -> dict[str, Any]:
return {
"id": observation.id,
"offer_id": observation.offer_id,
"source_id": observation.source_id,
"external_observation_id": observation.external_observation_id,
"observed_at": observation.observed_at,
"ingested_at": observation.ingested_at,
"item_price": observation.item_price,
"shipping_price": observation.shipping_price,
"currency": observation.currency,
"availability": observation.availability,
"raw_payload": observation.raw_payload,
"confidence": observation.confidence,
"total_price": observation.total_price,
}