fraudtwin.simulation#

Simulation scenario composition and generation controls.

Status: Stable

Classes#

fraudtwin.simulation.BehaviorDataset

Stable, ordered behavior profiles and their generated payment stream.

fraudtwin.simulation.BehaviorGenerator

Generate customer profiles and their payment stream from M1 entities.

fraudtwin.simulation.EntityDataset

Stable, ordered collections of all generated entities.

fraudtwin.simulation.EntityGenerator

Generate a reproducible population from a validated configuration.

fraudtwin.simulation.FraudDataset

Complete payment stream plus explainable M6 truth records.

fraudtwin.simulation.FraudScenarioGenerator

Add only explicit F01-F05 campaigns to an existing legitimate stream.

fraudtwin.simulation.FraudWorkflowDataset

Stable M7 workflow records derived from M6 fraud records.

fraudtwin.simulation.FraudWorkflowGenerator

Turn scenario-linked M6 records into a delayed operational truth path.

fraudtwin.simulation.GraphCampaign

Oracle descriptor for one deterministic scenario instance.

fraudtwin.simulation.GraphCampaignMembership

Oracle membership for one generated graph-fraud campaign.

fraudtwin.simulation.GraphEvidence

Normalized lineage for a derived relationship.

fraudtwin.simulation.GraphFraudDataset

GraphFraudDataset(payments: tuple[fraudtwin.domain.payments.Payment, ...], payment_events: tuple[fraudtwin.domain.payments.PaymentEvent, ...], ledger_entries: tuple[fraudtwin.domain.payments.LedgerEntry, ...], fraud_records: tuple[fraudtwin.domain.fraud.FraudRecord, ...], memberships: tuple[fraudtwin.domain.graph.GraphCampaignMembership, ...], patterns: tuple[fraudtwin.domain.graph.GraphPattern, ...], campaigns: tuple[fraudtwin.domain.graph.GraphCampaign, ...] = (), evidence: tuple[fraudtwin.domain.graph.GraphEvidence, ...] = (), hyperedges: tuple[fraudtwin.domain.graph.GraphHyperedge, ...] = (), hyperedge_memberships: tuple[fraudtwin.domain.graph.GraphHyperedgeMembership, ...] = (), dynamic: 'DynamicCampaignDataset | None' = None)

fraudtwin.simulation.GraphFraudGenerator

Generate graph scenarios using canonical participant allocation.

fraudtwin.simulation.GraphHyperedge

Optional higher-order campaign incidence record.

fraudtwin.simulation.GraphHyperedgeMembership

Membership of an entity in a higher-order graph relationship.

fraudtwin.simulation.GraphPattern

A deterministic structural pattern descriptor.

fraudtwin.simulation.NetworkEndpoint

An opt-in synthetic network endpoint; absent from legacy runs.

fraudtwin.simulation.PaymentDataset

Stable, ordered payment business objects and their event envelopes.

fraudtwin.simulation.PaymentGenerator

Generate positive, relationship-valid, legitimate payment events.

fraudtwin.simulation.QualityFaultInjector

Apply independently configurable, reproducible M8 faults.

fraudtwin.simulation.ResolvedDifficulty

Complete, hashable M12 difficulty configuration.

fraudtwin.simulation.ScenarioDifficultyPlan

Resolved transformations for one fraud or graph scenario.

Functions#

fraudtwin.simulation.apply_difficulty

Apply resolved controls to one scenario without changing its objective.

fraudtwin.simulation.count_pix_lifecycle_events

Count the explicit PIX lifecycle event types.

fraudtwin.simulation.generate_behavior

Convenience function for deterministic behavior generation.

fraudtwin.simulation.generate_entities

Convenience function for deterministic entity generation.

fraudtwin.simulation.resolve_difficulty

Resolve a level and its optional per-dimension replacements.

Detailed API#

Deterministic entity generation and batch output adapters.

class fraudtwin.simulation.BehaviorDataset(profiles, payments, payment_events, ledger_entries=(), fraud_records=(), alerts=(), fraud_cases=(), case_confirmations=(), customer_disputes=(), fraud_labels=(), label_observations=(), final_observed_labels=(), quality_fault_counts=<factory>, quality_fault_rates=<factory>, quality_diagnostics=<factory>, graph_memberships=(), graph_campaigns=(), graph_patterns=(), graph_evidence=(), graph_hyperedges=(), graph_hyperedge_memberships=(), camouflage_metadata=<factory>, oracle_tables=<factory>, quality_raw_faults=<factory>, schema_evolution_rows=<factory>, counterfactual=None, campaign_dynamics=None)[source][source]

Bases: object

Stable, ordered behavior profiles and their generated payment stream.

Parameters:

  • profiles (tuple[BehaviorProfile, ...])

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

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

  • 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, ...])

  • label_observations (tuple[LabelObservation, ...])

  • final_observed_labels (tuple[FinalObservedLabel, ...])

  • quality_fault_counts (dict[str, int])

  • quality_fault_rates (dict[str, float])

  • quality_diagnostics (dict[str, object])

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

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

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

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

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

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

  • camouflage_metadata (dict[str, object])

  • oracle_tables (dict[str, tuple[BaseModel, ...]])

  • quality_raw_faults (tuple[dict[str, object], ...])

  • schema_evolution_rows (dict[str, tuple[dict[str, object], ...]])

  • counterfactual (CounterfactualDataset | None)

  • campaign_dynamics (DynamicCampaignDataset | None)

property card_lifecycle_event_counts: dict[str, int]

Return counts for the explicit card event vocabulary.

property event_counts: dict[str, int]

Return payment, lifecycle, ledger, and fraud counts for the manifest.

property fraud_counts: dict[str, int]

Return manifest fraud counters, or empty counters when disabled.

property fraud_events: tuple[PaymentEvent, ...]

Return scenario-linked payment events, excluding hard negatives.

property fraud_rates: dict[str, float]

Return realized true-record rates by scenario.

property fraud_record_counts: dict[str, int]

Count true scenario records for manifest reporting.

property pix_lifecycle_event_counts: dict[str, int]

Return counts for the explicit PIX event vocabulary.

tables()[source][source]

Return all behavior tables in their stable export order.

Return type:

dict[str, tuple[BaseModel, ...]]

class fraudtwin.simulation.BehaviorGenerator(config, entities, simulation_run_id=None, calibration=None, stage_timings=None)[source][source]

Bases: object

Generate customer profiles and their payment stream from M1 entities.

Parameters:

  • config (SimulationRunConfig)

  • entities (EntityDataset)

  • simulation_run_id (str | None)

  • calibration (ResolvedCalibration | None)

  • stage_timings (dict[str, dict[str, float]] | None)

generate()[source][source]

Generate profiles, base payments, fraud scenarios, and events.

Return type:

BehaviorDataset

generate_profiles()[source][source]

Generate one deterministic profile per existing customer.

Return type:

tuple[BehaviorProfile, ...]

class fraudtwin.simulation.EntityDataset(customers, institutions, accounts, cards, merchants, devices, pix_keys, state_history=(), network_endpoints=())[source][source]

Bases: object

Stable, ordered collections of all generated entities.

Parameters:

  • customers (tuple[Customer, ...])

  • institutions (tuple[Institution, ...])

  • accounts (tuple[Account, ...])

  • cards (tuple[Card, ...])

  • merchants (tuple[Merchant, ...])

  • devices (tuple[Device, ...])

  • pix_keys (tuple[PixKey, ...])

  • state_history (tuple[EntityStateChange, ...])

  • network_endpoints (tuple[NetworkEndpoint, ...])

property all_ids: frozenset[str]

Return every string identifier carried by the entity collections.

all_tables()[source][source]

Return base entity tables plus optional effective-dated history.

Return type:

dict[str, tuple[Account | Card | Customer | Device | EntityStateChange | Institution | Merchant | PixKey | NetworkEndpoint, ...]]

property counts: dict[str, int]

Return output counts using the population configuration names.

property reference_ids: dict[str, frozenset[str]]

Return the stable IDs used by relationship validators.

tables()[source][source]

Return entities keyed by their stable output table names.

Return type:

dict[str, tuple[Account | Card | Customer | Device | EntityStateChange | Institution | Merchant | PixKey | NetworkEndpoint, ...]]

class fraudtwin.simulation.EntityGenerator(config, calibration=None)[source][source]

Bases: object

Generate a reproducible population from a validated configuration.

Parameters:
generate()[source][source]

Generate all entities in dependency order with isolated streams.

Return type:

EntityDataset

class fraudtwin.simulation.FraudDataset(payments, payment_events, ledger_entries, fraud_records)[source][source]

Bases: object

Complete payment stream plus explainable M6 truth records.

Parameters:

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

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

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

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

class fraudtwin.simulation.FraudScenarioGenerator(config, accounts, cards, merchants, devices, pix_keys, baseline, simulation_run_id=None)[source][source]

Bases: object

Add only explicit F01-F05 campaigns to an existing legitimate stream.

Parameters:

  • config (SimulationRunConfig)

  • accounts (tuple[Account, ...])

  • cards (tuple[Card, ...])

  • merchants (tuple[Merchant, ...])

  • devices (tuple[Device, ...])

  • pix_keys (tuple[PixKey, ...])

  • baseline (PaymentDataset)

  • simulation_run_id (str | None)

generate()[source][source]

Generate deterministic campaigns and preserve baseline output when disabled.

Return type:

FraudDataset

class fraudtwin.simulation.FraudWorkflowDataset(alerts, cases, confirmations, disputes, labels)[source][source]

Bases: object

Stable M7 workflow records derived from M6 fraud records.

Parameters:

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

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

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

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

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

class fraudtwin.simulation.FraudWorkflowGenerator(config, entities, fraud_dataset)[source][source]

Bases: object

Turn scenario-linked M6 records into a delayed operational truth path.

Parameters:

  • config (SimulationRunConfig)

  • entities (EntityDataset)

  • fraud_dataset (FraudDataset)

generate()[source][source]

Generate one deterministic workflow per selected alert/case path.

Return type:

FraudWorkflowDataset

class fraudtwin.simulation.PaymentDataset(payments, payment_events, ledger_entries=())[source][source]

Bases: object

Stable, ordered payment business objects and their event envelopes.

Parameters:

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

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

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

class fraudtwin.simulation.PaymentGenerator(config, accounts, cards, merchants, devices, pix_keys=(), simulation_run_id=None, calibration=None, include_lifecycle=True, include_ledger=True)[source][source]

Bases: object

Generate positive, relationship-valid, legitimate payment events.

Parameters:

  • config (SimulationRunConfig)

  • accounts (tuple[Account, ...])

  • cards (tuple[Card, ...])

  • merchants (tuple[Merchant, ...])

  • devices (tuple[Device, ...])

  • pix_keys (tuple[PixKey, ...])

  • simulation_run_id (str | None)

  • calibration (ResolvedCalibration | None)

  • include_lifecycle (bool)

  • include_ledger (bool)

generate(profiles)[source][source]

Materialize the deterministic payment stream as an in-memory dataset.

Return type:

PaymentDataset

Parameters:

profiles (tuple[BehaviorProfile, ...])

iter_generate(profiles)[source][source]

Stream exactly daily_target * duration_days legitimate payment pairs.

Return type:

Iterator[tuple[Payment, PaymentEvent]]

Parameters:

profiles (tuple[BehaviorProfile, ...])

materialize_ledger(payments, events)[source][source]

Reconcile a complete payment stream, including scenario payments.

Return type:

tuple[LedgerEntry, ...]

Parameters:

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

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

fraudtwin.simulation.count_pix_lifecycle_events(events)[source][source]

Count the explicit PIX lifecycle event types.

Return type:

dict[str, int]

Parameters:

events (Iterable[PaymentEvent])

class fraudtwin.simulation.QualityFaultInjector(config)[source][source]

Bases: object

Apply independently configurable, reproducible M8 faults.

Parameters:

config (SimulationRunConfig)

apply(dataset)[source][source]

Return a quality-mutated dataset and measured fault metadata.

Return type:

BehaviorDataset

Parameters:

dataset (BehaviorDataset)

fraudtwin.simulation.generate_behavior(config, entities)[source][source]

Convenience function for deterministic behavior generation.

Return type:

BehaviorDataset

Parameters:

  • config (SimulationRunConfig)

  • entities (EntityDataset)

fraudtwin.simulation.generate_entities(config)[source][source]

Convenience function for deterministic entity generation.

Return type:

EntityDataset

Parameters:

config (SimulationRunConfig)

class fraudtwin.simulation.GraphCampaignMembership(**data)[source][source]

Bases: _EntityModel

Oracle membership for one generated graph-fraud campaign.

Parameters:

  • campaign_id (str)

  • pattern_type (Literal['MULE_NETWORK', 'CYCLIC_RING', 'BENEFICIARY_NETWORK', 'SHARED_DEVICE_INFRASTRUCTURE', 'SHARED_IP_INFRASTRUCTURE', 'FAN_IN', 'FAN_OUT', 'SHORT_MONEY_DWELL', 'DENSE_CAMPAIGN', 'MERCHANT_CUSTOMER_COMMUNITY', 'BIPARTITE_NETWORK', 'STACKED_NETWORK', 'SCATTER_GATHER', 'GATHER_SCATTER', 'RANDOM_ALERT_CONTROL'])

  • member_id (str)

  • member_type (str)

  • role (str)

  • valid_from (datetime)

  • valid_to (datetime | None)

  • source_event_id (str | None)

  • payment_id (str | 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.simulation.GraphCampaign(**data)[source][source]

Bases: _EntityModel

Oracle descriptor for one deterministic scenario instance.

Parameters:

  • campaign_id (str)

  • scenario_type (Literal['MULE_NETWORK', 'CYCLIC_RING', 'BENEFICIARY_NETWORK', 'SHARED_DEVICE_INFRASTRUCTURE', 'SHARED_IP_INFRASTRUCTURE', 'FAN_IN', 'FAN_OUT', 'SHORT_MONEY_DWELL', 'DENSE_CAMPAIGN', 'MERCHANT_CUSTOMER_COMMUNITY', 'BIPARTITE_NETWORK', 'STACKED_NETWORK', 'SCATTER_GATHER', 'GATHER_SCATTER', 'RANDOM_ALERT_CONTROL'])

  • scenario_code (str)

  • truth_label (Literal['FRAUD', 'CONTROL'])

  • valid_from (datetime)

  • valid_to (datetime)

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

  • modifiers (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.simulation.GraphPattern(**data)[source][source]

Bases: _EntityModel

A deterministic structural pattern descriptor.

Parameters:

  • pattern_id (str)

  • pattern_type (Literal['MULE_NETWORK', 'CYCLIC_RING', 'BENEFICIARY_NETWORK', 'SHARED_DEVICE_INFRASTRUCTURE', 'SHARED_IP_INFRASTRUCTURE', 'FAN_IN', 'FAN_OUT', 'SHORT_MONEY_DWELL', 'DENSE_CAMPAIGN', 'MERCHANT_CUSTOMER_COMMUNITY', 'BIPARTITE_NETWORK', 'STACKED_NETWORK', 'SCATTER_GATHER', 'GATHER_SCATTER', 'RANDOM_ALERT_CONTROL'])

  • campaign_id (str | None)

  • detected_at (datetime)

  • window_from (datetime)

  • window_to (datetime)

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

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

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

  • threshold (Annotated[int | None, Ge(ge=1)])

  • observed_value (float | None)

  • invariant_status (Literal['PASS', 'UNAVAILABLE', 'FAIL'])

  • truth_label (Literal['FRAUD', 'CONTROL'])

  • scenario_code (str | None)

  • reason (str | 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.simulation.GraphEvidence(**data)[source][source]

Bases: _EntityModel

Normalized lineage for a derived relationship.

Parameters:

  • evidence_id (str)

  • edge_id (str | None)

  • evidence_type (str)

  • resource_id (str | None)

  • source_event_id (str | None)

  • payment_id (str | None)

  • observed_at (datetime)

  • available_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.simulation.GraphHyperedge(**data)[source][source]

Bases: _EntityModel

Optional higher-order campaign incidence record.

Parameters:

  • hyperedge_id (str)

  • hyperedge_type (Literal['STRUCTURAL', 'SEMANTIC'])

  • campaign_id (str | None)

  • pattern_id (str | None)

  • valid_from (datetime)

  • valid_to (datetime | None)

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

  • payment_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.simulation.GraphHyperedgeMembership(**data)[source][source]

Bases: _EntityModel

Membership of an entity in a higher-order graph relationship.

Parameters:

  • hyperedge_id (str)

  • member_id (str)

  • member_type (str)

  • role (str | 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.simulation.NetworkEndpoint(**data)[source][source]

Bases: _EntityModel

An opt-in synthetic network endpoint; absent from legacy runs.

Parameters:

  • endpoint_id (str)

  • endpoint_type (Literal['IP', 'NETWORK'])

  • address_hash (str)

  • first_seen_at (datetime)

  • last_seen_at (datetime)

  • valid_from (datetime)

  • valid_to (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.simulation.GraphFraudDataset(payments, payment_events, ledger_entries, fraud_records, memberships, patterns, campaigns=(), evidence=(), hyperedges=(), hyperedge_memberships=(), dynamic=None)[source][source]

Bases: object

Parameters:

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

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

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

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

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

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

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

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

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

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

  • dynamic (DynamicCampaignDataset | None)

class fraudtwin.simulation.GraphFraudGenerator(config, accounts, devices, endpoints, baseline, simulation_run_id=None, merchants=(), pix_keys=())[source][source]

Bases: object

Generate graph scenarios using canonical participant allocation.

Parameters:

  • config (SimulationRunConfig)

  • accounts (tuple[Account, ...])

  • devices (tuple[Device, ...])

  • endpoints (tuple[NetworkEndpoint, ...])

  • baseline (PaymentDataset | FraudDataset)

  • simulation_run_id (str | None)

  • merchants (tuple[Merchant, ...])

  • pix_keys (tuple[PixKey, ...])

class fraudtwin.simulation.ResolvedDifficulty(**data)[source][source]

Bases: BaseModel

Complete, hashable M12 difficulty configuration.

Parameters:

  • enabled (bool)

  • resolver_version (str)

  • requested_difficulty (int | None)

  • requested_controls (dict[str, float | None])

  • resolved_controls (dict[str, float])

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

  • effective_configuration_hash (str)

property level: int | None

Return the requested level for convenient API consumers.

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

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

plan(scenario)[source][source]

Return the immutable plan for one scenario.

Return type:

ScenarioDifficultyPlan

Parameters:

scenario (str)

class fraudtwin.simulation.ScenarioDifficultyPlan(**data)[source][source]

Bases: BaseModel

Resolved transformations for one fraud or graph scenario.

Parameters:

  • scenario (str)

  • amount_similarity (float)

  • behavior_similarity (float)

  • merchant_similarity (float)

  • device_similarity (float)

  • channel_similarity (float)

  • geography_similarity (float)

  • scenario_subtlety (float)

  • noise_hard_negatives (float)

  • prevalence (float)

  • temporal_irregularity (float)

  • graph_structural_subtlety (float)

  • amount_multiplier (float)

  • timing_multiplier (float)

  • hard_negative_multiplier (float)

  • parameter_transformations (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].

fraudtwin.simulation.resolve_difficulty(config)[source][source]

Resolve a level and its optional per-dimension replacements.

Return type:

ResolvedDifficulty

Parameters:

config (SimulationRunConfig)

fraudtwin.simulation.apply_difficulty(resolved, scenario, context=None)[source][source]

Apply resolved controls to one scenario without changing its objective.

context is accepted for extension compatibility and is intentionally excluded from the plan hash; all stochastic choices use generator-owned M12 stream names.

Return type:

ScenarioDifficultyPlan

Parameters: