Core contracts¶

Configuration¶

Central configuration loading and deterministic resolution.

class p2p.config.AggregateRun(path, labels)[source]¶

Bases: object

One resolved aggregation input and its condition labels.

Parameters:

path (Path)
labels (Mapping[str, Any])

path: Path¶

labels: Mapping[str, Any]¶

to_dict()[source]¶

Return type:: dict[str, Any]

class p2p.config.AggregateSpec(runs, group_by, metrics, expected_repeats, expected_cells, require_complete, center, spread, hierarchy_reduce, hierarchy_weighting, missing_policy, missing_reason, export_formats, export_precision)[source]¶

Bases: object

Validated aggregation configuration without execution logic.

Parameters:

runs (tuple[AggregateRun, ...])
group_by (tuple[str, ...])
metrics (tuple[str, ...])
expected_repeats (int | None)
expected_cells (tuple[Mapping[str, Any], ...])
require_complete (bool)
center (str)
spread (str)
hierarchy_reduce (tuple[str, ...])
hierarchy_weighting (str)
missing_policy (str)
missing_reason (str | None)
export_formats (tuple[str, ...])
export_precision (int)

runs: tuple[AggregateRun, ...]¶

group_by: tuple[str, ...]¶

metrics: tuple[str, ...]¶

expected_repeats: int | None¶

expected_cells: tuple[Mapping[str, Any], ...]¶

require_complete: bool¶

center: str¶

spread: str¶

hierarchy_reduce: tuple[str, ...]¶

hierarchy_weighting: str¶

missing_policy: str¶

missing_reason: str | None¶

export_formats: tuple[str, ...]¶

export_precision: int¶

to_dict()[source]¶

Return type:: dict[str, Any]

class p2p.config.Config(source_path, data, attribute_bank=None, aggregate_spec=None)[source]¶

Bases: object

Immutable resolved central configuration.

Parameters:

source_path (Path)
data (Mapping[str, Any])
attribute_bank (ResolvedAttributeBank | None)
aggregate_spec (AggregateSpec | None)

source_path: Path¶

data: Mapping[str, Any]¶

attribute_bank: ResolvedAttributeBank | None¶

aggregate_spec: AggregateSpec | None¶

classmethod load(path)[source]¶

Parameters:: path (str | PathLike[str])
Return type:: Config

classmethod from_mapping(mapping, *, declared_in)[source]¶

Parameters:

mapping (Mapping[str, Any])
declared_in (str | PathLike[str])

Return type:

Config

property digest: str¶: Return a stable digest of the resolved configuration.

write_snapshot(run_dir)[source]¶

Write the resolved configuration snapshot.

Parameters:: run_dir (str | PathLike[str])
Return type:: Path

validate_for(verb)[source]¶

Validate the authoritative verb requirement matrix.

Parameters:: verb (str)
Return type:: Config

to_dict()[source]¶

Return type:: dict[str, Any]

exception p2p.config.ConfigError[source]¶

Bases: ValueError

Base error for central configuration failures.

exception p2p.config.ConfigLoadError[source]¶

Bases: ConfigError

Raised when a manifest cannot be loaded or inherited.

exception p2p.config.ConfigValidationError[source]¶

Bases: ConfigError

Raised when resolved configuration data is invalid.

class p2p.config.ResolvedAttributeBank(source, source_digest, data)[source]¶

Bases: object

Immutable validated attribute-bank source.

Parameters:

source (str)
source_digest (str)
data (Mapping[str, Any])

source: str¶

source_digest: str¶

data: Mapping[str, Any]¶

to_dict()[source]¶

Return type:: dict[str, Any]

write_snapshot(run_dir, *, attribute_bank=None)[source]¶

Parameters:

run_dir (str | PathLike[str])
attribute_bank (Mapping[str, Any] | None)

Return type:

Path

p2p.config.expand_grid(spec)[source]¶

Expand a literal list, linspace, or base-10 logspace grid.

Parameters:: spec (Any)
Return type:: tuple[Any, …]

Earlier-format configuration conversion¶

One-off migration of relevant legacy P2P configurations.

exception p2p.config_conversion.ConfigConversionError[source]¶

Bases: ConfigError

Raised when a legacy configuration cannot be migrated safely.

p2p.config_conversion.convert_earlier_config(source, destination, *, family, source_root=None, method=None, legacy_root=None)[source]¶

Convert one earlier-format YAML configuration into a central manifest.

source_root is the root that the earlier configuration’s relative paths are resolved against. legacy_root is its earlier name and is still accepted.

Parameters:

source (str | PathLike[str])
destination (str | PathLike[str])
family (str)
source_root (str | PathLike[str] | None)
method (str | None)
legacy_root (str | PathLike[str] | None)

Return type:

Path

Registries¶

Central registries for swappable P2P components.

class p2p.registry.Registry(registry_name)[source]¶

Bases: Generic[FactoryT]

Named collection of factories with duplicate protection.

Parameters:: registry_name (str)

register(name, factory)[source]¶

Parameters:

name (str)
factory (FactoryT)

Return type:

None

get(name)[source]¶

Return a registered factory.

Parameters:: name (str)
Return type:: FactoryT

create(name, **kwargs)[source]¶

Instantiate a registered factory using keyword arguments.

Parameters:

name (str)
kwargs (Any)

Return type:

Any

names()[source]¶

Return registered names in deterministic order.

Return type:: tuple[str, …]

Metrics¶

Pure numerical metrics and explicit probability transformations.

p2p.metrics.clip_and_normalize(values, *, qid=None)[source]¶

Clip negative entries and normalize the remaining positive mass.

Parameters:

values (ArrayLike)
qid (str | None)

Return type:

NDArray[float64]

p2p.metrics.emd(observed, predicted, *, qid=None, atol=1e-08)[source]¶

Return normalized EMD for vectors in canonical option order.

Parameters:

observed (ArrayLike)
predicted (ArrayLike)
qid (str | None)
atol (float)

Return type:

float

p2p.metrics.mcpa(observed, predicted, *, qid=None, atol=1e-08)[source]¶

Return majority-category prediction accuracy for one question.

Parameters:

observed (ArrayLike)
predicted (ArrayLike)
qid (str | None)
atol (float)

Return type:

float

p2p.metrics.mse(observed, predicted)[source]¶

Return raw option-level mean squared error without transformation.

Parameters:

observed (ArrayLike)
predicted (ArrayLike)

Return type:

float

p2p.metrics.one_minus_jsd(observed, predicted, *, qid=None, atol=1e-08)[source]¶

Return one minus Jensen-Shannon distance using natural logarithms.

Parameters:

observed (ArrayLike)
predicted (ArrayLike)
qid (str | None)
atol (float)

Return type:

float

p2p.metrics.project_to_simplex(values, *, qid=None)[source]¶

Return the Euclidean projection onto the unit probability simplex.

Parameters:

values (ArrayLike)
qid (str | None)

Return type:

NDArray[float64]

p2p.metrics.validate_distribution(values, *, name='distribution', qid=None, atol=1e-08)[source]¶

Validate and return a probability vector without transforming it.

Parameters:

values (ArrayLike)
name (str)
qid (str | None)
atol (float)

Return type:

NDArray[float64]

Canonical results¶

Canonical run provenance and long-form metric records.

class p2p.results.MetricArtifact(run_id, metrics)[source]¶

Bases: object

Canonical metrics.json payload for one run.

Parameters:

run_id (str)
metrics (Sequence[MetricRecord])

run_id: str¶

metrics: Sequence[MetricRecord]¶

schema_version: int¶

to_dict()[source]¶

Return type:: dict[str, Any]

class p2p.results.MetricRecord(metric, value, split, model, unit, prediction_policy, qids_requested, qids_used, diagnostics=<factory>)[source]¶

Bases: object

One canonical long-form metric observation.

Parameters:

metric (str)
value (float)
split (str)
model (str)
unit (str)
prediction_policy (str)
qids_requested (int)
qids_used (int)
diagnostics (Mapping[str, Any])

metric: str¶

value: float¶

split: str¶

model: str¶

unit: str¶

prediction_policy: str¶

qids_requested: int¶

qids_used: int¶

diagnostics: Mapping[str, Any]¶

property identity: tuple[str, ...]¶: Fields identifying one metric row.

to_dict()[source]¶

Return type:: dict[str, Any]

class p2p.results.RunRecord(run_id, repeat, seed, conditions, config_digest, cleaning_policy, source_artifacts)[source]¶

Bases: object

Canonical provenance stored in one run.json artifact.

Parameters:

run_id (str)
repeat (int)
seed (int)
conditions (Mapping[str, Any])
config_digest (str)
cleaning_policy (str)
source_artifacts (Mapping[str, SourceArtifact])

run_id: str¶

repeat: int¶

seed: int¶

conditions: Mapping[str, Any]¶

config_digest: str¶

cleaning_policy: str¶

source_artifacts: Mapping[str, SourceArtifact]¶

schema_version: int¶

to_dict()[source]¶

Return type:: dict[str, Any]

class p2p.results.SourceArtifact(path, digest)[source]¶

Bases: object

One source path and its captured SHA-256 digest.

Parameters:

path (str)
digest (str)

path: str¶

digest: str¶

to_dict()[source]¶

Return type:: dict[str, str]

Aggregation¶

Pure validation and reduction of canonical single-run artifacts.

class p2p.aggregation.AggregationArtifacts(resolved_spec, inputs_csv, validation_json, runs_csv, summary_csv, summary_json, markdown=None, latex=None)[source]¶

Bases: object

Files emitted for one validated aggregation.

Parameters:

resolved_spec (Path)
inputs_csv (Path)
validation_json (Path)
runs_csv (Path)
summary_csv (Path)
summary_json (Path)
markdown (Path | None)
latex (Path | None)

resolved_spec: Path¶

inputs_csv: Path¶

validation_json: Path¶

runs_csv: Path¶

summary_csv: Path¶

summary_json: Path¶

markdown: Path | None¶

latex: Path | None¶

property required_files: tuple[Path, ...]¶

class p2p.aggregation.AggregationResult(input_rows, run_rows, summary_rows, validation, aggregation_digest)[source]¶

Bases: object

Validated full-precision rows produced by the pure core.

Parameters:

input_rows (tuple[dict[str, Any], ...])
run_rows (tuple[dict[str, Any], ...])
summary_rows (tuple[dict[str, Any], ...])
validation (Mapping[str, Any])
aggregation_digest (str)

input_rows: tuple[dict[str, Any], ...]¶

run_rows: tuple[dict[str, Any], ...]¶

summary_rows: tuple[dict[str, Any], ...]¶

validation: Mapping[str, Any]¶

aggregation_digest: str¶

exception p2p.aggregation.AggregationValidationError[source]¶

Bases: ValueError

Raised when canonical aggregation inputs are incompatible.

p2p.aggregation.aggregate_runs(spec)[source]¶

Validate and reduce canonical runs without writing files.

Parameters:: spec (AggregateSpec | None)
Return type:: AggregationResult

p2p.aggregation.write_aggregation_artifacts(result, spec, *, output_dir, source_config=None)[source]¶

Write ledgers and full-precision summary artifacts.

Parameters:

result (AggregationResult)
spec (AggregateSpec)
output_dir (str | PathLike[str])
source_config (str | PathLike[str] | None)

Return type:

AggregationArtifacts

Sweeps¶

Generic deterministic sweep expansion.

class p2p.sweep.SweepSpec(kind, target_verb, overrides, repeats=1)[source]¶

Bases: object

Normalized sweep metadata over a resolved base config.

Parameters:

kind (str)
target_verb (str)
overrides (tuple[Mapping[str, Any], ...])
repeats (int)

kind: str¶

target_verb: str¶

overrides: tuple[Mapping[str, Any], ...]¶

repeats: int¶

classmethod from_config(config, kind)[source]¶

Normalize one ablate, panel, or simulation sweep.

Parameters:

config (Config)
kind (str)

Return type:

SweepSpec

p2p.sweep.expand_sweep(base, spec)[source]¶

Expand overrides into resolved target-validated configs.

Parameters:

base (Config)
spec (SweepSpec)

Return type:

tuple[Config, …]

Usage accounting¶

Backend-neutral token usage and cost accounting.

class p2p.usage.TokenPricing(input_per_million_usd=0.0, output_per_million_usd=0.0, cached_input_per_million_usd=0.0)[source]¶

Bases: object

USD pricing per one million tokens.

Parameters:

input_per_million_usd (float)
output_per_million_usd (float)
cached_input_per_million_usd (float)

input_per_million_usd: float¶

output_per_million_usd: float¶

cached_input_per_million_usd: float¶

class p2p.usage.UsageArtifacts(csv_path, json_path)[source]¶

Bases: object

CSV and JSON usage artifacts for one command run.

Parameters:

csv_path (Path)
json_path (Path)

csv_path: Path¶

json_path: Path¶

class p2p.usage.UsageRecord(usage, model_name, module_name, agent_id, cost_usd)[source]¶

Bases: object

One accounted backend operation.

Parameters:

usage (TokenUsage)
model_name (str)
module_name (str)
agent_id (str)
cost_usd (float)

usage: TokenUsage¶

model_name: str¶

module_name: str¶

agent_id: str¶

cost_usd: float¶

to_dict()[source]¶

Return a JSON-serializable usage record.

Return type:: dict[str, str | int | float]

class p2p.usage.UsageTracker(pricing=None, *, stream_path=None, resume_stream=False)[source]¶

Bases: object

Accumulate normalized usage with required model/module metadata.

Parameters:

pricing (Mapping[str, TokenPricing] | None)
stream_path (str | os.PathLike[str] | None)
resume_stream (bool)

log(*, usage, model_name, module_name, agent_id)[source]¶

Account for one normalized backend response exactly once.

Parameters:

usage (TokenUsage)
model_name (str)
module_name (str)
agent_id (str)

Return type:

UsageRecord

property records: tuple[UsageRecord, ...]¶: Return an immutable snapshot of accounted operations.

summary()[source]¶

Return full-precision aggregate token counts and cost.

Return type:: dict[str, int | float]

report()[source]¶

Return a compact human-readable usage summary.

Return type:: str

p2p.usage.write_usage_artifacts(tracker, run_dir)[source]¶

Write full-precision normalized usage records and totals.

Parameters:

tracker (UsageTracker)
run_dir (str | PathLike[str])

Return type:

UsageArtifacts

Budget forecasting¶

Historical token and cost forecasting for API-backed P2P runs.

class p2p.budgeting.BudgetArtifacts(json_path, csv_path, history_csv_path)[source]¶

Bases: object

Files written for one forecast.

Parameters:

json_path (Path)
csv_path (Path)
history_csv_path (Path)

json_path: Path¶

csv_path: Path¶

history_csv_path: Path¶

class p2p.budgeting.BudgetFeatures(target_n, question_count, survey_chars, option_count, initial_n, update_steps, batch_size, patch_fraction, patch_top_k, max_tokens, is_active, is_vanilla)[source]¶

Bases: object

Numerical run features used by the forecasting model.

Parameters:

target_n (int)
question_count (int)
survey_chars (int)
option_count (int)
initial_n (int)
update_steps (int)
batch_size (int)
patch_fraction (float)
patch_top_k (int)
max_tokens (int)
is_active (int)
is_vanilla (int)

target_n: int¶

question_count: int¶

survey_chars: int¶

option_count: int¶

initial_n: int¶

update_steps: int¶

batch_size: int¶

patch_fraction: float¶

patch_top_k: int¶

max_tokens: int¶

is_active: int¶

is_vanilla: int¶

vector()[source]¶

Return features in the stable model order.

Return type:: list[float]

class p2p.budgeting.BudgetForecast(schema_version, artifact_type, config_source, config_digest, target_command, confidence, history_roots, history_digest, historical_runs, historical_records, features, scenarios, history_rows, pilot_run, pilot_records, notes)[source]¶

Bases: object

Complete reproducible forecast and model-card payload.

Parameters:

schema_version (int)
artifact_type (str)
config_source (str)
config_digest (str)
target_command (str)
confidence (float)
history_roots (tuple[str, ...])
history_digest (str)
historical_runs (int)
historical_records (int)
features (BudgetFeatures)
scenarios (tuple[ScenarioBudget, ...])
history_rows (tuple[dict[str, Any], ...])
pilot_run (str | None)
pilot_records (int)
notes (tuple[str, ...])

schema_version: int¶

artifact_type: str¶

config_source: str¶

config_digest: str¶

target_command: str¶

confidence: float¶

history_roots: tuple[str, ...]¶

history_digest: str¶

historical_runs: int¶

historical_records: int¶

features: BudgetFeatures¶

scenarios: tuple[ScenarioBudget, ...]¶

history_rows: tuple[dict[str, Any], ...]¶

pilot_run: str | None¶

pilot_records: int¶

notes: tuple[str, ...]¶

to_dict()[source]¶

Return a JSON-compatible payload.

Return type:: dict[str, Any]

class p2p.budgeting.IntervalEstimate(low, point, high)[source]¶

Bases: object

Point and central uncertainty interval.

Parameters:

low (float)
point (float)
high (float)

low: float¶

point: float¶

high: float¶

class p2p.budgeting.ModuleBudget(module_name, model_name, backend_type, training_scope, training_runs, coverage, calls, input_tokens, output_tokens, cached_input_tokens, cost_usd, pricing, pilot_calibration, oob_median_absolute_percentage_error)[source]¶

Bases: object

Forecast for one API-using pipeline module.

Parameters:

module_name (str)
model_name (str)
backend_type (str)
training_scope (str)
training_runs (int)
coverage (str)
calls (IntervalEstimate)
input_tokens (IntervalEstimate)
output_tokens (IntervalEstimate)
cached_input_tokens (IntervalEstimate)
cost_usd (IntervalEstimate | None)
pricing (PricingEstimate)
pilot_calibration (dict[str, float] | None)
oob_median_absolute_percentage_error (dict[str, float] | None)

module_name: str¶

model_name: str¶

backend_type: str¶

training_scope: str¶

training_runs: int¶

coverage: str¶

calls: IntervalEstimate¶

input_tokens: IntervalEstimate¶

output_tokens: IntervalEstimate¶

cached_input_tokens: IntervalEstimate¶

cost_usd: IntervalEstimate | None¶

pricing: PricingEstimate¶

pilot_calibration: dict[str, float] | None¶

oob_median_absolute_percentage_error: dict[str, float] | None¶

class p2p.budgeting.PricingEstimate(model_name, input_per_million_usd, output_per_million_usd, cached_input_per_million_usd, source)[source]¶

Bases: object

One resolved per-million-token price with provenance.

Parameters:

model_name (str)
input_per_million_usd (float | None)
output_per_million_usd (float | None)
cached_input_per_million_usd (float | None)
source (str)

model_name: str¶

input_per_million_usd: float | None¶

output_per_million_usd: float | None¶

cached_input_per_million_usd: float | None¶

source: str¶

property available: bool¶

class p2p.budgeting.ScenarioBudget(name, modules, total_tokens, total_cost_usd, budget_usd, budget_status)[source]¶

Bases: object

Forecast for one configured or comparison model scenario.

Parameters:

name (str)
modules (tuple[ModuleBudget, ...])
total_tokens (IntervalEstimate)
total_cost_usd (IntervalEstimate | None)
budget_usd (float | None)
budget_status (str)

name: str¶

modules: tuple[ModuleBudget, ...]¶

total_tokens: IntervalEstimate¶

total_cost_usd: IntervalEstimate | None¶

budget_usd: float | None¶

budget_status: str¶

p2p.budgeting.collect_budget_history(roots)[source]¶

Collect one model/module row per historical tracked run.

Parameters:: roots (Sequence[str | PathLike[str]])
Return type:: tuple[DataFrame, str, tuple[str, …]]

p2p.budgeting.forecast_budget(config, *, history_roots, target_command='pipeline', comparison_models=(), pricing_path=None, pilot_run=None, confidence=0.8, budget_usd=None)[source]¶

Train historical models and forecast one configured API run.

Parameters:

config (Config)
history_roots (Sequence[str | PathLike[str]])
target_command (str)
comparison_models (Sequence[str])
pricing_path (str | PathLike[str] | None)
pilot_run (str | PathLike[str] | None)
confidence (float)
budget_usd (float | None)

Return type:

BudgetForecast

p2p.budgeting.load_pricing(path)[source]¶

Load explicit current prices from a small provider-neutral YAML.

Parameters:: path (str | PathLike[str] | None)
Return type:: dict[str, PricingEstimate]

p2p.budgeting.write_budget_artifacts(forecast, output_dir)[source]¶

Write canonical JSON/CSV forecast and training provenance.

Parameters:

forecast (BudgetForecast)
output_dir (str | PathLike[str])

Return type:

BudgetArtifacts

Self-contained interactive presentation for budget forecasts.

class p2p.budget_dashboard.BudgetDashboard(output_path, sections)[source]¶

Bases: object

One generated interactive budget report.

Parameters:

output_path (Path)
sections (tuple[str, ...])

output_path: Path¶

sections: tuple[str, ...]¶

p2p.budget_dashboard.build_budget_dashboard(forecast, output_path, *, refresh_seconds=None)[source]¶

Write one self-contained interactive budget dashboard.

Parameters:

forecast (BudgetForecast)
output_path (str | PathLike[str])
refresh_seconds (float | None)

Return type:

BudgetDashboard

Offline entropy sampling¶

Deterministic, offline sampling over an existing response basis.

class p2p.sampling.EntropySample(selected_eids, target_entropy, achieved_entropy, absolute_error, sample_size, method, candidates_evaluated, seed)[source]¶

Bases: object

Best subset found for an explicitly requested mean entropy.

Parameters:

selected_eids (tuple[str, ...])
target_entropy (float)
achieved_entropy (float)
absolute_error (float)
sample_size (int)
method (str)
candidates_evaluated (int)
seed (int)

selected_eids: tuple[str, ...]¶

target_entropy: float¶

achieved_entropy: float¶

absolute_error: float¶

sample_size: int¶

method: str¶

candidates_evaluated: int¶

seed: int¶

to_dict()[source]¶

Return type:: dict[str, Any]

p2p.sampling.sample_to_target_mean_entropy(endowments, responses, survey, *, sample_size, target_entropy, seed=101, max_evaluations=5000, tolerance=None, require_tolerance=False)[source]¶

Select a fixed-size subset closest to a requested mean entropy.

The objective calls the ActiveEndowments.compute_entropy() kernel, so the estimand is the same mean of normalized per-question categorical entropies that active generation tracks. Small search spaces are enumerated exactly. Larger ones use a seeded random search followed by deterministic one-swap refinement. The global random state is never touched.

Parameters:

sample_size (int)
target_entropy (float)
seed (int)
max_evaluations (int)
tolerance (float | None)
require_tolerance (bool)

Return type:

EntropySample