fraudtwin.domain.campaign_dynamics#

Domain types for evolving fraud campaigns.

Status: Stable

Classes#

fraudtwin.domain.campaign_dynamics.CampaignActorMembershipChange

fraudtwin.domain.campaign_dynamics.CampaignIntensityDecision

fraudtwin.domain.campaign_dynamics.CampaignLineage

fraudtwin.domain.campaign_dynamics.CampaignPhaseChange

fraudtwin.domain.campaign_dynamics.CampaignSourceSnapshot

fraudtwin.domain.campaign_dynamics.CampaignStateSnapshot

fraudtwin.domain.campaign_dynamics.CampaignTopologyMutation

fraudtwin.domain.campaign_dynamics.CampaignTransition

Detailed API#

Immutable domain records for campaign evolution.

class fraudtwin.domain.campaign_dynamics.CampaignActorMembershipChange(**data)[source][source]

Bases: _EntityModel

Parameters:

  • membership_change_id (str)

  • campaign_id (str)

  • actor_id (str)

  • actor_type (Literal['ACCOUNT', 'CUSTOMER', 'CARD', 'DEVICE', 'MERCHANT'])

  • role (str)

  • action (Literal['JOIN', 'LEAVE', 'ROTATE_IN', 'ROTATE_OUT'])

  • occurred_at (datetime)

  • valid_from (datetime)

  • valid_to (datetime | None)

  • source_entity_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.domain.campaign_dynamics.CampaignIntensityDecision(**data)[source][source]

Bases: _EntityModel

Parameters:

  • intensity_decision_id (str)

  • campaign_id (str)

  • phase (Literal['compromise', 'setup', 'transfer', 'cash_out', 'dormant', 'closed'])

  • decision_at (datetime)

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

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

  • model_name (str)

  • stream_id (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.campaign_dynamics.CampaignLineage(**data)[source][source]

Bases: _EntityModel

Parameters:

  • lineage_id (str)

  • campaign_id (str)

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

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

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

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

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

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

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

  • derived_at (datetime)

  • reason (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.campaign_dynamics.CampaignPhaseChange(**data)[source][source]

Bases: _EntityModel

Parameters:

  • phase_change_id (str)

  • campaign_id (str)

  • transition_id (str)

  • phase (Literal['compromise', 'setup', 'transfer', 'cash_out', 'dormant', 'closed'])

  • 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.domain.campaign_dynamics.CampaignSourceSnapshot(**data)[source][source]

Bases: _EntityModel

Parameters:

  • source_snapshot_id (str)

  • campaign_id (str)

  • source_run_id (str)

  • captured_at (datetime)

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

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

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

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

  • configuration_hash (str)

  • schema_fingerprint (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.campaign_dynamics.CampaignStateSnapshot(**data)[source][source]

Bases: _EntityModel

Parameters:

  • snapshot_id (str)

  • campaign_id (str)

  • phase (Literal['compromise', 'setup', 'transfer', 'cash_out', 'dormant', 'closed'])

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

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

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

  • valid_from (datetime)

  • valid_to (datetime | None)

  • transition_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.domain.campaign_dynamics.CampaignTopologyMutation(**data)[source][source]

Bases: _EntityModel

Parameters:

  • topology_mutation_id (str)

  • campaign_id (str)

  • mutation_type (Literal['MULE_ROTATION', 'DEVICE_ROTATION', 'RING_SPLIT', 'RING_MERGE', 'CROSS_RAIL', 'HYPEREDGE_ADD'])

  • occurred_at (datetime)

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

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

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

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

  • reason (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.campaign_dynamics.CampaignTransition(**data)[source][source]

Bases: _EntityModel

Parameters:

  • transition_id (str)

  • campaign_id (str)

  • from_phase (Literal['compromise', 'setup', 'transfer', 'cash_out', 'dormant', 'closed'])

  • to_phase (Literal['compromise', 'setup', 'transfer', 'cash_out', 'dormant', 'closed'])

  • occurred_at (datetime)

  • reason (str)

  • model_name (str)

  • source_snapshot_id (str | None)

  • stream_id (str)

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

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