fraudtwin.graph#

Temporal graph construction, validation, and export.

Status: Stable

Classes#

`fraudtwin.graph.GraphCampaign`	Oracle descriptor for one deterministic scenario instance.
`fraudtwin.graph.GraphDataset`	One immutable graph view and its structural descriptors.
`fraudtwin.graph.GraphEdge`	Time-bounded relationship with provenance back to source events.
`fraudtwin.graph.GraphEvidence`	Normalized lineage for a derived relationship.
`fraudtwin.graph.GraphHyperedge`	Optional higher-order campaign incidence record.
`fraudtwin.graph.GraphHyperedgeMembership`	Membership of an entity in a higher-order graph relationship.
`fraudtwin.graph.GraphNode`	Versioned entity node in an observable or oracle graph view.

Functions#

`fraudtwin.graph.build_graph`	Build a deterministic temporal graph without generating source data.
`fraudtwin.graph.to_pyg`	Convert a graph to `torch_geometric.data.HeteroData` when installed.
`fraudtwin.graph.validate_graph`	Validate IDs, temporal closure, and edge references for one graph view.
`fraudtwin.graph.validate_graph_scenarios`	Validate campaign closure and higher-order incidence records.
`fraudtwin.graph.write_graph`	Write one append-only graph artifact containing the requested views.

Constants and protocols#

Name	Reference
`GRAPH_CAMPAIGN_SCHEMA`	`fraudtwin.graph.GRAPH_CAMPAIGN_SCHEMA`
`GRAPH_EDGE_SCHEMA`	`fraudtwin.graph.GRAPH_EDGE_SCHEMA`
`GRAPH_EVIDENCE_SCHEMA`	`fraudtwin.graph.GRAPH_EVIDENCE_SCHEMA`
`GRAPH_HYPEREDGE_MEMBERSHIP_SCHEMA`	`fraudtwin.graph.GRAPH_HYPEREDGE_MEMBERSHIP_SCHEMA`
`GRAPH_HYPEREDGE_SCHEMA`	`fraudtwin.graph.GRAPH_HYPEREDGE_SCHEMA`
`GRAPH_MEMBERSHIP_SCHEMA`	`fraudtwin.graph.GRAPH_MEMBERSHIP_SCHEMA`
`GRAPH_NODE_SCHEMA`	`fraudtwin.graph.GRAPH_NODE_SCHEMA`
`GRAPH_PATTERN_SCHEMA`	`fraudtwin.graph.GRAPH_PATTERN_SCHEMA`
`GRAPH_SCHEMA_VERSION`	`fraudtwin.graph.GRAPH_SCHEMA_VERSION`

Detailed API#

Deterministic graph views over one generated FraudTwin run.

class fraudtwin.graph.GraphDataset(view, source_run_id, nodes, edges, memberships, patterns, from_time, to_time, as_of, campaigns=(), evidence=(), hyperedges=(), hyperedge_memberships=())[source][source]

Bases: object

One immutable graph view and its structural descriptors.

Parameters:

view (Literal['observable', 'oracle'])
source_run_id (str)
nodes (tuple[GraphNode, ...])
edges (tuple[GraphEdge, ...])
memberships (tuple[GraphCampaignMembership, ...])
patterns (tuple[GraphPattern, ...])
from_time (datetime)
to_time (datetime)
as_of (datetime)
campaigns (tuple[GraphCampaign, ...])
evidence (tuple[GraphEvidence, ...])
hyperedges (tuple[GraphHyperedge, ...])
hyperedge_memberships (tuple[GraphHyperedgeMembership, ...])

class fraudtwin.graph.GraphEdge(**data)[source][source]

Bases: BaseModel

Time-bounded relationship with provenance back to source events.

Parameters:

edge_id (str)
src_id (str)
dst_id (str)
edge_type (str)
event_time (datetime | None)
valid_from (datetime)
valid_to (datetime | None)
source_event_id (str | None)
payment_id (str | None)
scenario_id (str | None)
campaign_id (str | None)
edge_origin (Literal['DOMAIN_EVENT', 'DERIVED_RELATION', 'SCENARIO_GROUND_TRUTH'])
weight (float | None)
support_count (int | None)
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.graph.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.graph.GraphNode(**data)[source][source]

Bases: BaseModel

Versioned entity node in an observable or oracle graph view.

Parameters:

node_id (str)
node_type (str)
valid_from (datetime)
valid_to (datetime | None)
available_at (datetime | None)
attributes_version (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.graph.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.graph.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.graph.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].

fraudtwin.graph.build_graph(config, entities, behavior, source_manifest=None, *, view='observable', as_of=None, from_time=None, to_time=None, memberships=(), evidence=(), hyperedges=(), hyperedge_memberships=(), campaigns=())[source][source]

Build a deterministic temporal graph without generating source data.

Return type:

GraphDataset

Parameters:

config (SimulationRunConfig)
entities (EntityDataset)
behavior (BehaviorDataset)
source_manifest (RunManifest | None)
view (Literal['observable', 'oracle'])
as_of (datetime | None)
from_time (datetime | None)
to_time (datetime | None)
memberships (tuple[GraphCampaignMembership, ...])
evidence (tuple[GraphEvidence, ...])
hyperedges (tuple[GraphHyperedge, ...])
hyperedge_memberships (tuple[GraphHyperedgeMembership, ...])
campaigns (tuple[GraphCampaign, ...])

fraudtwin.graph.to_pyg(dataset)[source][source]

Convert a graph to torch_geometric.data.HeteroData when installed.

Return type:: Any
Parameters:: dataset (GraphDataset)

fraudtwin.graph.validate_graph(dataset)[source][source]

Validate IDs, temporal closure, and edge references for one graph view.

Return type:: None
Parameters:: dataset (GraphDataset)

fraudtwin.graph.validate_graph_scenarios(dataset)[source][source]

Validate campaign closure and higher-order incidence records.

Return type:: None
Parameters:: dataset (GraphDataset)

fraudtwin.graph.write_graph(datasets, output_dir, *, source_manifest, config, formats=('parquet', 'neo4j'))[source][source]

Write one append-only graph artifact containing the requested views.

Return type:

tuple[Path, GraphManifest]

Parameters:

datasets (dict[Literal['observable', 'oracle'], ~fraudtwin.graph.GraphDataset])
output_dir (Path)
source_manifest (RunManifest)
config (SimulationRunConfig)
formats (tuple[str, ...])