fraudtwin.scale#

Partition planning, checkpoints, and scale iteration.

Status: Experimental

Classes#

`fraudtwin.scale.ChunkCompletion`	Fingerprint and location for one atomically completed chunk.
`fraudtwin.scale.ChunkDescriptor`	Half-open logical ordinal range belonging to a stable shard.
`fraudtwin.scale.PartitionCompletion`	Fingerprint and counts for one atomically completed partition.
`fraudtwin.scale.ReconciliationResult`	Cross-partition invariant results recorded in manifests/checkpoints.
`fraudtwin.scale.ScaleCheckpoint`	Serializable checkpoint manifest for deterministic resume.
`fraudtwin.scale.ScalePlan`	Resolved, immutable execution parameters for one scale run.
`fraudtwin.scale.ShardDescriptor`	Stable shard identity; it is independent of the worker executing it.

Functions#

`fraudtwin.scale.aggregate_fingerprint`	Hash rows incrementally without retaining the complete input.
`fraudtwin.scale.checkpoint_fingerprint`	Return a stable fingerprint excluding the wall-clock creation time.
`fraudtwin.scale.chunk_payment_ranges`	Yield stable shard-local payment chunks without materializing IDs.
`fraudtwin.scale.create_scale_stream_rng`	Create a deterministic hierarchical stream for one scale task.
`fraudtwin.scale.fingerprint_rows`	Hash canonical JSON rows, independent of process or file ordering.
`fraudtwin.scale.iter_chunks`	Yield deterministic half-open chunk descriptors.
`fraudtwin.scale.iter_partition_query`	Execute an out-of-core DuckDB query over partitioned Parquet.
`fraudtwin.scale.iter_partition_rows`	Read scale chunks one file at a time, keeping reader memory bounded.
`fraudtwin.scale.iter_partition_table`	Yield rows for one logical table from partitioned scale output.
`fraudtwin.scale.iter_payment_ranges`	Yield deterministic contiguous payment ordinal ranges per shard.
`fraudtwin.scale.load_checkpoint`	Load and validate a checkpoint manifest.
`fraudtwin.scale.partition_id`	Return the canonical shard ID for a logical record.
`fraudtwin.scale.partition_index`	Map a logical ID to a stable shard without using Python hash randomization.
`fraudtwin.scale.reconcile_logical_ids`	Detect duplicate/lost IDs and return a serializable reconciliation result.
`fraudtwin.scale.require_scale_plan`	Resolve an enabled scale plan and narrow away the optional result.
`fraudtwin.scale.resolve_scale_plan`	Resolve the opt-in scale configuration, returning `None` when disabled.
`fraudtwin.scale.run_scale_benchmark`	Run one explicitly requested scale job and publish machine evidence.
`fraudtwin.scale.shard_descriptors`	Return shard descriptors in canonical index order.
`fraudtwin.scale.write_checkpoint`	Atomically publish a checkpoint manifest.
`fraudtwin.scale.write_scale_benchmark_manifest`	Write non-deterministic machine evidence separately from run identity.
`fraudtwin.scale.write_scale_partitions`	Write deterministic partition/chunk artifacts and a checkpoint.

Constants and protocols#

Name	Reference
`SEED_TREE_VERSION`	`fraudtwin.scale.SEED_TREE_VERSION`

Detailed API#

Deterministic chunk execution, sharding, and reconciliation for M18.

The scale layer consumes canonical row iterators and owns partitioning, bounded-memory spooling, persistence, and checkpoint metadata. It does not contain a second simulation model.

class fraudtwin.scale.ScalePlan(**data)[source][source]

Bases: BaseModel

Resolved, immutable execution parameters for one scale run.

Parameters:

profile (Literal['dev', 'small', 'medium', 'large', 'xlarge', 'billion'])
target_logical_events (int)
shard_count (Annotated[int, Ge(ge=1)])
chunk_size (Annotated[int, Ge(ge=1)])
worker_count (Annotated[int, Ge(ge=1)])
output_batch_size (Annotated[int, Ge(ge=1)])
checkpoint_frequency_chunks (Annotated[int, Ge(ge=1)])
partition_mapping (Literal['stable_hash_v1'])
seed (Annotated[int, Ge(ge=0)])
run_id (str)
configuration_hash (str)
target_payments (Annotated[int, Ge(ge=0)])
seed_tree_version (str)
features (tuple[Literal['entities', 'behavior', 'payments', 'lifecycle', 'ledger', 'fraud', 'labels', 'graph', 'pit', 'backtest'], ...])
storage_backend (Literal['local', 'fsspec'])
storage_uri (str | None)
state_backend (Literal['duckdb'])
manifest_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.scale.ShardDescriptor(**data)[source][source]

Bases: BaseModel

Stable shard identity; it is independent of the worker executing it.

Parameters:

shard_index (Annotated[int, Ge(ge=0)])
shard_id (str)
mapping (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.scale.ChunkDescriptor(**data)[source][source]

Bases: BaseModel

Half-open logical ordinal range belonging to a stable shard.

Parameters:

shard_id (str)
chunk_index (Annotated[int, Ge(ge=0)])
start_ordinal (Annotated[int, Ge(ge=0)])
end_ordinal (Annotated[int, Gt(gt=0)])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class fraudtwin.scale.PartitionCompletion(**data)[source][source]

Bases: BaseModel

Fingerprint and counts for one atomically completed partition.

Parameters:

shard_id (str)
row_count (Annotated[int, Ge(ge=0)])
chunk_count (Annotated[int, Ge(ge=0)])
fingerprint (str)
checksum (str)
path (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.scale.ChunkCompletion(**data)[source][source]

Bases: BaseModel

Fingerprint and location for one atomically completed chunk.

Parameters:

shard_id (str)
logical_type (str)
chunk_index (Annotated[int, Ge(ge=0)])
start_ordinal (Annotated[int, Ge(ge=0)])
end_ordinal (Annotated[int, Ge(ge=0)])
row_count (Annotated[int, Ge(ge=0)])
fingerprint (str)
checksum (str)
path (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.scale.ReconciliationResult(**data)[source][source]

Bases: BaseModel

Cross-partition invariant results recorded in manifests/checkpoints.

Parameters:

logical_row_count (Annotated[int, Ge(ge=0)])
partition_row_count (Annotated[int, Ge(ge=0)])
duplicate_logical_ids (tuple[str, ...])
missing_logical_ids (tuple[str, ...])
ledger_debit_total (float)
ledger_credit_total (float)
ledger_balanced (bool)
shard_ledger_totals (dict[str, dict[str, float]])
transfer_reconciliation_records (Annotated[int, Ge(ge=0)])
account_balance_violations (Annotated[int, Ge(ge=0)])
accounts_reconciled (Annotated[int, Ge(ge=0)])
valid (bool)
checks (dict[str, bool])

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class fraudtwin.scale.ScaleCheckpoint(**data)[source][source]

Bases: BaseModel

Serializable checkpoint manifest for deterministic resume.

Parameters:

checkpoint_version (str)
run_id (str)
run_dir (str)
configuration_hash (str)
resolved_configuration (dict[str, Any])
seed (Annotated[int, Ge(ge=0)])
seed_tree_version (str)
plan (ScalePlan)
completed_partitions (tuple[PartitionCompletion, ...])
completed_chunks (tuple[ChunkCompletion, ...])
reconciliation (ReconciliationResult | None)
created_at (datetime)
integrity_hash (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.scale.resolve_scale_plan(config, *, run_id=None)[source][source]

Resolve the opt-in scale configuration, returning None when disabled.

Return type:

ScalePlan | None

Parameters:

config (SimulationRunConfig)
run_id (str | None)

fraudtwin.scale.require_scale_plan(config, *, run_id=None)[source][source]

Resolve an enabled scale plan and narrow away the optional result.

Use this helper in applications that require scale execution. It turns a disabled scale section into a clear configuration error and gives static type checkers a concrete ScalePlan for shard and checkpoint code.

Raises:

ValueError – If scale generation is not enabled in config.

Return type:

ScalePlan

Parameters:

config (SimulationRunConfig)
run_id (str | None)

fraudtwin.scale.shard_descriptors(plan)[source][source]

Return shard descriptors in canonical index order.

Return type:: tuple[ShardDescriptor, ...]
Parameters:: plan (ScalePlan)

fraudtwin.scale.partition_index(logical_id, shard_count)[source][source]

Map a logical ID to a stable shard without using Python hash randomization.

Return type:

int

Parameters:

logical_id (str)
shard_count (int)

fraudtwin.scale.partition_id(logical_id, shard_count)[source][source]

Return the canonical shard ID for a logical record.

Return type:

str

Parameters:

logical_id (str)
shard_count (int)

fraudtwin.scale.create_scale_stream_rng(seed, *components)[source][source]

Create a deterministic hierarchical stream for one scale task.

Return type:

Random

Parameters:

seed (int)
components (str)

fraudtwin.scale.iter_chunks(shard_id, start, end, chunk_size)[source][source]

Yield deterministic half-open chunk descriptors.

Return type:

Iterator[ChunkDescriptor]

Parameters:

shard_id (str)
start (int)
end (int)
chunk_size (int)

fraudtwin.scale.fingerprint_rows(rows)[source][source]

Hash canonical JSON rows, independent of process or file ordering.

Return type:: str
Parameters:: rows (Iterable[Mapping[str, Any]])

fraudtwin.scale.aggregate_fingerprint(rows)[source][source]

Hash rows incrementally without retaining the complete input.

Return type:: str
Parameters:: rows (Iterable[Mapping[str, Any]])

fraudtwin.scale.iter_payment_ranges(target_payments, shard_count)[source][source]

Yield deterministic contiguous payment ordinal ranges per shard.

Return type:

Iterator[tuple[str, int, int, int]]

Parameters:

target_payments (int)
shard_count (int)

fraudtwin.scale.chunk_payment_ranges(target_payments, shard_count, chunk_size)[source][source]

Yield stable shard-local payment chunks without materializing IDs.

Return type:

Iterator[tuple[str, int, int, int]]

Parameters:

target_payments (int)
shard_count (int)
chunk_size (int)

fraudtwin.scale.iter_partition_rows(run_dir, *, shard_id=None, columns=None)[source][source]

Read scale chunks one file at a time, keeping reader memory bounded.

Return type:

Iterator[dict[str, Any]]

Parameters:

run_dir (str | Path)
shard_id (str | None)
columns (list[str] | None)

fraudtwin.scale.iter_partition_table(run_dir, logical_type, *, shard_id=None, columns=None)[source][source]

Yield rows for one logical table from partitioned scale output.

This is intentionally lazy and keeps the existing in-memory readers untouched. Consumers that can operate out-of-core should use this helper instead of loading a complete generated run.

Return type:

Iterator[dict[str, Any]]

Parameters:

run_dir (str | Path)
logical_type (str)
shard_id (str | None)
columns (list[str] | None)

fraudtwin.scale.iter_partition_query(run_dir, query, *, shard_id=None, batch_size=10_000)[source][source]

Execute an out-of-core DuckDB query over partitioned Parquet.

query must reference {source}, which is replaced by a read_parquet relation. Results are yielded in bounded record batches; DuckDB is imported lazily so the core generator has no dependency on it.

Return type:

Iterator[dict[str, Any]]

Parameters:

run_dir (str | Path)
query (str)
shard_id (str | None)
batch_size (int)

fraudtwin.scale.write_scale_benchmark_manifest(path, *, target_payments, realized_counts, configuration_hash, seed, shard_count, worker_count, elapsed_seconds, resume=None, output_dir=None, claim_scope='laptop-dev-only', command=None, stage_timings=None)[source][source]

Write non-deterministic machine evidence separately from run identity.

Return type:

Path

Parameters:

path (str | Path)
target_payments (int)
realized_counts (Mapping[str, int])
configuration_hash (str)
seed (int)
shard_count (int)
worker_count (int)
elapsed_seconds (float)
resume (Mapping[str, Any] | None)
output_dir (str | Path | None)
claim_scope (str)
command (str | None)
stage_timings (Mapping[str, Mapping[str, float]] | None)

fraudtwin.scale.run_scale_benchmark(config, *, output_dir, checkpoint_dir=None, evidence_dir=None, command=None)[source][source]

Run one explicitly requested scale job and publish machine evidence.

This is intentionally a manual-job API. It does not change the normal generation path or make large profiles part of unit/CI tests.

Return type:

Path

Parameters:

config (SimulationRunConfig)
output_dir (str | Path)
checkpoint_dir (str | Path | None)
evidence_dir (str | Path | None)
command (str | None)

fraudtwin.scale.reconcile_logical_ids(logical_ids, partition_ids, *, expected_ids=None)[source][source]

Detect duplicate/lost IDs and return a serializable reconciliation result.

Return type:

ReconciliationResult

Parameters:

logical_ids (Iterable[str])
partition_ids (Iterable[str])
expected_ids (Iterable[str] | None)

fraudtwin.scale.write_checkpoint(path, checkpoint)[source][source]

Atomically publish a checkpoint manifest.

Return type:

Path

Parameters:

path (str | Path)
checkpoint (ScaleCheckpoint)

fraudtwin.scale.load_checkpoint(path)[source][source]

Load and validate a checkpoint manifest.

Return type:: ScaleCheckpoint
Parameters:: path (str | Path)

fraudtwin.scale.write_scale_partitions(run_dir, plan, records, *, checkpoint_dir=None, resolved_configuration=None, stage_timings=None)[source][source]

Write deterministic partition/chunk artifacts and a checkpoint.

Records are expected to contain a unique logical_id and logical_type. An optional partition_key keeps account-local records together; worker execution is intentionally not represented in the output contract.

Return type:

tuple[tuple[PartitionCompletion, ...], ReconciliationResult, Path]

Parameters:

run_dir (str | Path)
plan (ScalePlan)
records (Iterable[Mapping[str, Any]])
checkpoint_dir (str | Path | None)
resolved_configuration (dict[str, Any] | None)
stage_timings (dict[str, dict[str, float]] | None)

fraudtwin.scale.checkpoint_fingerprint(checkpoint)[source][source]

Return a stable fingerprint excluding the wall-clock creation time.

Return type:: str
Parameters:: checkpoint (ScaleCheckpoint)