fraudtwin.domain.cases#

Reusable domain case and scenario records.

Status: Stable

Classes#

fraudtwin.domain.cases.CustomerDispute

A customer-submitted dispute event with the common event envelope.

fraudtwin.domain.cases.DelayedFraudLabel

A fraud label exposed only after the configured operational evidence.

fraudtwin.domain.cases.FraudAlert

An automated alert causally raised from one M6 fraud record.

fraudtwin.domain.cases.FraudCase

An investigation opened from an alert and its delayed label state.

fraudtwin.domain.cases.FraudCaseConfirmation

A case decision, recorded separately from the fraud ground truth.

Functions#

fraudtwin.domain.cases.validate_fraud_workflow

Validate M7 references, causal chains, and temporal availability rules.

Constants and protocols#

Name

Reference

FraudWorkflowEventType

fraudtwin.domain.cases.FraudWorkflowEventType

InvestigationOutcome

fraudtwin.domain.cases.InvestigationOutcome

Detailed API#

Fraud investigations and operationally delayed labels.

class fraudtwin.domain.cases.CustomerDispute(**data)[source][source]

Bases: _EntityModel

A customer-submitted dispute event with the common event envelope.

Parameters:

  • event_id (str)

  • event_type (Literal['CUSTOMER_DISPUTE_SUBMITTED'])

  • event_version (Annotated[int, Ge(ge=1)])

  • fraud_case_id (str)

  • fraud_alert_id (str)

  • fraud_record_id (str)

  • underlying_event_id (str)

  • payment_id (str)

  • customer_id (str)

  • account_id (str | None)

  • card_id (str | None)

  • device_id (str | None)

  • merchant_id (str | None)

  • event_time (datetime)

  • source_created_at (datetime)

  • source_available_at (datetime)

  • ingested_at (datetime)

  • processed_at (datetime)

  • producer (str)

  • source_system (str)

  • schema_version (str)

  • correlation_id (str)

  • causation_id (str)

  • simulation_run_id (str | None)

  • scenario_id (str)

  • scenario_type (str)

  • payment_rail (str)

  • payment_type (str)

  • amount (Annotated[float, Gt(gt=0)])

  • currency (str)

  • affected_entity_ids (tuple[str, ...])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class fraudtwin.domain.cases.DelayedFraudLabel(**data)[source][source]

Bases: _EntityModel

A fraud label exposed only after the configured operational evidence.

Parameters:

  • label_id (str)

  • fraud_case_id (str)

  • fraud_alert_id (str)

  • fraud_record_id (str)

  • customer_id (str)

  • payment_id (str)

  • event_id (str)

  • scenario_id (str)

  • scenario_type (str)

  • label (Literal['FRAUD', 'LEGITIMATE'])

  • fraud_truth (bool | None)

  • fraud_occurred_at (datetime)

  • fraud_confirmed_at (datetime | None)

  • dispute_event_at (datetime | None)

  • label_available_at (datetime)

  • investigation_outcome (Literal['CONFIRMED_FRAUD', 'FALSE_POSITIVE', 'CUSTOMER_DISPUTE', 'UNRESOLVED', 'LEGITIMATE'])

  • amount (Annotated[float, Gt(gt=0)])

  • currency (str)

  • correlation_id (str)

  • causation_id (str)

  • simulation_run_id (str | None)

  • affected_entity_ids (tuple[str, ...])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class fraudtwin.domain.cases.FraudAlert(**data)[source][source]

Bases: _EntityModel

An automated alert causally raised from one M6 fraud record.

Parameters:

  • fraud_alert_id (str)

  • alert_type (Literal['AUTOMATED_SCENARIO_ALERT'])

  • severity (Literal['LOW', 'MEDIUM', 'HIGH'])

  • customer_id (str)

  • account_id (str | None)

  • card_id (str | None)

  • device_id (str | None)

  • merchant_id (str | None)

  • payment_id (str)

  • event_id (str)

  • fraud_record_id (str)

  • scenario_id (str)

  • scenario_type (str)

  • trigger (str)

  • reason (str)

  • alert_created_at (datetime)

  • amount (Annotated[float, Gt(gt=0)])

  • currency (str)

  • correlation_id (str)

  • causation_id (str)

  • simulation_run_id (str | None)

  • affected_entity_ids (tuple[str, ...])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class fraudtwin.domain.cases.FraudCase(**data)[source][source]

Bases: _EntityModel

An investigation opened from an alert and its delayed label state.

Parameters:

  • fraud_case_id (str)

  • fraud_alert_id (str)

  • customer_id (str)

  • account_id (str | None)

  • card_id (str | None)

  • device_id (str | None)

  • merchant_id (str | None)

  • payment_id (str)

  • event_id (str)

  • fraud_record_id (str)

  • scenario_id (str)

  • scenario_type (str)

  • fraud_truth (bool | None)

  • fraud_occurred_at (datetime)

  • alert_created_at (datetime)

  • case_opened_at (datetime)

  • case_closed_at (datetime | None)

  • fraud_confirmed_at (datetime | None)

  • label_available_at (datetime | None)

  • investigation_outcome (Literal['CONFIRMED_FRAUD', 'FALSE_POSITIVE', 'CUSTOMER_DISPUTE', 'UNRESOLVED', 'LEGITIMATE'])

  • loss_amount (Annotated[float, Ge(ge=0)])

  • recovered_amount (Annotated[float, Ge(ge=0)])

  • amount (Annotated[float, Gt(gt=0)])

  • currency (str)

  • correlation_id (str)

  • causation_id (str)

  • simulation_run_id (str | None)

  • affected_entity_ids (tuple[str, ...])

  • case_reopened_at (datetime | None)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class fraudtwin.domain.cases.FraudCaseConfirmation(**data)[source][source]

Bases: _EntityModel

A case decision, recorded separately from the fraud ground truth.

Parameters:

  • confirmation_id (str)

  • fraud_case_id (str)

  • fraud_alert_id (str)

  • customer_id (str)

  • payment_id (str)

  • event_id (str)

  • fraud_record_id (str)

  • scenario_id (str)

  • scenario_type (str)

  • fraud_truth (bool | None)

  • confirmed_at (datetime)

  • investigation_outcome (Literal['CONFIRMED_FRAUD', 'FALSE_POSITIVE', 'CUSTOMER_DISPUTE', 'UNRESOLVED', 'LEGITIMATE'])

  • amount (Annotated[float, Gt(gt=0)])

  • currency (str)

  • correlation_id (str)

  • causation_id (str)

  • simulation_run_id (str | None)

  • affected_entity_ids (tuple[str, ...])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

fraudtwin.domain.cases.validate_fraud_workflow(alerts, cases, confirmations, disputes, labels, *, customer_ids=frozenset(), account_ids=frozenset(), card_ids=frozenset(), device_ids=frozenset(), merchant_ids=frozenset(), payment_ids=frozenset(), event_ids=frozenset(), fraud_record_ids=frozenset(), all_entity_ids=frozenset(), fraud_truth_by_record=None, fraud_records_by_id=None)[source][source]

Validate M7 references, causal chains, and temporal availability rules.

Return type:

None

Parameters:

  • alerts (tuple[FraudAlert, ...])

  • cases (tuple[FraudCase, ...])

  • confirmations (tuple[FraudCaseConfirmation, ...])

  • disputes (tuple[CustomerDispute, ...])

  • labels (tuple[DelayedFraudLabel, ...])

  • customer_ids (frozenset[str])

  • account_ids (frozenset[str])

  • card_ids (frozenset[str])

  • device_ids (frozenset[str])

  • merchant_ids (frozenset[str])

  • payment_ids (frozenset[str])

  • event_ids (frozenset[str])

  • fraud_record_ids (frozenset[str])

  • all_entity_ids (frozenset[str])

  • fraud_truth_by_record (Mapping[str, bool] | None)

  • fraud_records_by_id (Mapping[str, FraudRecord] | None)