fraudtwin.counterfactual#

Counterfactual scenario generation and resolution.

Status: Experimental

Classes#

fraudtwin.counterfactual.CounterfactualChangeSet

Machine-readable explanation of one accepted or rejected request.

fraudtwin.counterfactual.CounterfactualDataset

Original/modified sidecar records and oracle metadata.

fraudtwin.counterfactual.CounterfactualScope

Resolved controls for one objective request.

fraudtwin.counterfactual.DistanceFunction

Protocol implemented by pluggable counterfactual distances.

fraudtwin.counterfactual.ResolvedCounterfactual

Complete immutable M14 configuration resolution.

fraudtwin.counterfactual.SourceTrajectory

One legitimate source payment and its point-in-time snapshot.

Functions#

fraudtwin.counterfactual.generate_counterfactuals

Generate deterministic sidecar counterfactuals from a pristine stream.

fraudtwin.counterfactual.register_distance_function

Register a deterministic named distance function.

fraudtwin.counterfactual.resolve_counterfactual

Resolve M14 precedence without consuming a random stream.

fraudtwin.counterfactual.select_source_trajectories

Select deterministic legitimate-only source trajectories.

Constants and protocols#

Name

Reference

ALL_OBJECTIVES

fraudtwin.counterfactual.ALL_OBJECTIVES

Detailed API#

Deterministic counterfactual generation.

The engine deliberately operates on a pristine legitimate payment stream. It produces append-only sidecar data and never mutates the source models or the ordinary output tables.

class fraudtwin.counterfactual.CounterfactualChangeSet(**data)[source][source]

Bases: BaseModel

Machine-readable explanation of one accepted or rejected request.

Parameters:

  • change_set_id (str)

  • request_index (int)

  • objective (str)

  • status (str)

  • source_payment_id (str | None)

  • derived_payment_id (str | None)

  • requested_budget (float)

  • resolved_budget (float)

  • effective_distance (float | None)

  • per_dimension_costs (dict[str, float])

  • changed_fields (dict[str, dict[str, object]])

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

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

  • objective_satisfaction (dict[str, object])

  • rejection_reason (str | None)

  • source_to_counterfactual (dict[str, 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.counterfactual.CounterfactualDataset(**data)[source][source]

Bases: BaseModel

Original/modified sidecar records and oracle metadata.

Parameters:

  • counterfactual_id (str)

  • original_payments (tuple[Payment, ...])

  • original_events (tuple[PaymentEvent, ...])

  • original_ledger_entries (tuple[LedgerEntry, ...])

  • modified_payments (tuple[Payment, ...])

  • modified_events (tuple[PaymentEvent, ...])

  • modified_ledger_entries (tuple[LedgerEntry, ...])

  • fraud_records (tuple[FraudRecord, ...])

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

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

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

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

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

  • change_sets (tuple[CounterfactualChangeSet, ...])

  • graph_campaigns (tuple[GraphCampaign, ...])

  • graph_memberships (tuple[GraphCampaignMembership, ...])

  • graph_patterns (tuple[GraphPattern, ...])

  • graph_evidence (tuple[GraphEvidence, ...])

  • graph_hyperedges (tuple[GraphHyperedge, ...])

  • graph_hyperedge_memberships (tuple[GraphHyperedgeMembership, ...])

  • metadata (dict[str, object])

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

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

class fraudtwin.counterfactual.CounterfactualScope(**data)[source][source]

Bases: BaseModel

Resolved controls for one objective request.

Parameters:

  • objective (Literal['F01', 'F02', 'F03', 'F04', 'F05', 'MULE_NETWORK', 'CYCLIC_RING', 'BENEFICIARY_NETWORK', 'FAN_OUT', 'BIPARTITE_NETWORK', 'STACKED_NETWORK', 'SCATTER_GATHER', 'GATHER_SCATTER', 'SHARED_DEVICE_INFRASTRUCTURE', 'SHARED_IP_INFRASTRUCTURE', 'DENSE_CAMPAIGN', 'MERCHANT_CUSTOMER_COMMUNITY', 'RANDOM_ALERT_CONTROL'])

  • family (str)

  • enabled (bool)

  • budget (float)

  • costs (dict[str, float])

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

  • distance_function (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.counterfactual.DistanceFunction(*args, **kwargs)[source][source]

Bases: Protocol

Protocol implemented by pluggable counterfactual distances.

class fraudtwin.counterfactual.ResolvedCounterfactual(**data)[source][source]

Bases: BaseModel

Complete immutable M14 configuration resolution.

Parameters:

  • enabled (bool)

  • resolver_version (str)

  • requested (dict[str, object])

  • scopes (tuple[CounterfactualScope, ...])

  • effective_configuration_hash (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.counterfactual.SourceTrajectory(**data)[source][source]

Bases: BaseModel

One legitimate source payment and its point-in-time snapshot.

Parameters:

  • source_payment (Payment)

  • source_events (tuple[PaymentEvent, ...])

  • source_ledger_entries (tuple[LedgerEntry, ...])

  • decision_at (datetime)

  • available_event_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.counterfactual.generate_counterfactuals(config, entities, baseline, *, counterfactual_id=None, run_id=None)[source][source]

Generate deterministic sidecar counterfactuals from a pristine stream.

Return type:

CounterfactualDataset

Parameters:

  • config (SimulationRunConfig)

  • entities (EntityDataset)

  • baseline (PaymentDataset)

  • counterfactual_id (str | None)

  • run_id (str | None)

fraudtwin.counterfactual.register_distance_function(name, function)[source][source]

Register a deterministic named distance function.

Return type:

None

Parameters:

  • name (str)

  • function (Callable[[Mapping[str, Mapping[str, object]], Mapping[str, float]], float])

fraudtwin.counterfactual.resolve_counterfactual(config)[source][source]

Resolve M14 precedence without consuming a random stream.

Return type:

ResolvedCounterfactual

Parameters:

config (SimulationRunConfig)

fraudtwin.counterfactual.select_source_trajectories(config, payments, *, events=None, ledger_entries=(), request_index=0)[source][source]

Select deterministic legitimate-only source trajectories.

Return type:

tuple[SourceTrajectory, ...]

Parameters:

  • config (SimulationRunConfig)

  • payments (tuple[Payment, ...] | PaymentDataset)

  • events (tuple[PaymentEvent, ...] | None)

  • ledger_entries (tuple[LedgerEntry, ...])

  • request_index (int)