OCDocker.OCScore.Optimization.Protocol module

Minimal staged protocol abstractions for OCScore optimization workflows.

It is imported as:

from OCDocker.OCScore.Optimization.Protocol import ProtocolContext

class OCDocker.OCScore.Optimization.Protocol.ProtocolContext(pdbbind_df, dudez_df, selected_features, output_dir, random_seed=42, metadata=<factory>, artifacts=<factory>, stage_results=<factory>, protocol_log=<factory>)[source]

Bases: object

Shared state passed between staged protocol steps.

Parameters:

  • pdbbind_df (pd.DataFrame) – Reduced PDBbind dataframe.

  • dudez_df (pd.DataFrame) – Reduced DUDEz dataframe.

  • selected_features (list[str]) – Descriptor columns selected by the feature-reduction protocol.

  • output_dir (str) – Directory where protocol artifacts are written.

  • random_seed (int, optional) – Random seed used by protocol stages, by default 42.

  • metadata (dict[str, Any], optional) – Optional dataset identifiers, paths, or caller metadata.

  • artifacts (dict[str, Any], optional) – Runtime artifacts passed between stages.

  • stage_results (dict[str, Any], optional) – JSON-serializable stage summaries.

  • protocol_log (dict[str, Any], optional) – Reproducibility log accumulated by the protocol.

pdbbind_df: DataFrame
dudez_df: DataFrame
selected_features: list[str]
output_dir: str
random_seed: int = 42
metadata: dict[str, Any]
artifacts: dict[str, Any]
stage_results: dict[str, Any]
protocol_log: dict[str, Any]
ensure_output_dir()[source]

Create and return the output directory.

Returns:

Protocol output directory.

Return type:

pathlib.Path

class OCDocker.OCScore.Optimization.Protocol.ProtocolStage(*args, **kwargs)[source]

Bases: Protocol

Protocol implemented by executable protocol stages.

name

Stable stage name used in logs and result dictionaries.

Type:

str

name: str
run(context)[source]

Run the stage and return an updated context.

Parameters:

context (ProtocolContext) – Current protocol context.

Returns:

Updated protocol context.

Return type:

ProtocolContext

class OCDocker.OCScore.Optimization.Protocol.ReplicaResult(replica_index, replica_name, seed, output_dir, success, context=None, summary=<factory>, error=None, failed_stage=None, traceback=None)[source]

Bases: object

Result for one replicated staged protocol execution.

Parameters:

  • replica_index (int) – Zero-based replica index.

  • replica_name (str) – Stable replica name, for example "replica_000".

  • seed (int) – Random seed used by this replica.

  • output_dir (str) – Replica-specific output directory.

  • success (bool) – Whether the replica completed all stages.

  • context (ProtocolContext | None, optional) – Final replica context for successful replicas, by default None.

  • summary (dict[str, Any], optional) – Per-replica summary row used in aggregate reports.

  • error (str | None, optional) – Error message for failed replicas, by default None.

  • failed_stage (str | None, optional) – Stage name active when the replica failed, by default None.

  • traceback (str | None, optional) – Formatted traceback for failed replicas, by default None.

replica_index: int
replica_name: str
seed: int
output_dir: str
success: bool
context: ProtocolContext | None = None
summary: dict[str, Any]
error: str | None = None
failed_stage: str | None = None
traceback: str | None = None
class OCDocker.OCScore.Optimization.Protocol.ReplicatedProtocolConfig(n_replicas=1, base_seed=None, replica_name_prefix='replica', continue_on_replica_failure=False, write_reports=True, write_protocol_log=True, replica_jobs=1, resume_completed=False)[source]

Bases: object

Configuration for replicated staged protocol execution.

A replica is one full staged Optuna/modeling execution. Replicas are not Optuna trials; each replica contains its own PDBbind Optuna study, transfer stage, and DUDEz Optuna study. Feature reduction is expected to run once before this protocol and is not repeated per replica.

Parameters:

  • n_replicas (int, optional) – Number of independent staged protocol executions, by default 1.

  • base_seed (int | None, optional) – Base random seed. Replica i uses base_seed + i. If None, the input context random seed is used.

  • replica_name_prefix (str, optional) – Prefix used to build replica names, by default "replica".

  • continue_on_replica_failure (bool, optional) – If True, failed replicas are recorded and later replicas continue. If False, the first failed replica raises an exception after reports are written, by default False.

  • write_reports (bool, optional) – If True, write replicas_summary.csv, replicas_summary.json, and replicas_protocol.json in the base output directory, by default True.

  • write_protocol_log (bool, optional) – If True, each replica writes its own protocol_log.json, by default True.

  • replica_jobs (int, optional) – Number of replicas to execute concurrently. Values greater than one use concurrent worker threads, by default 1.

  • resume_completed (bool, optional) – If True, reuse completed replica directories with a valid protocol log instead of rerunning them, by default False.

n_replicas: int = 1
base_seed: int | None = None
replica_name_prefix: str = 'replica'
continue_on_replica_failure: bool = False
write_reports: bool = True
write_protocol_log: bool = True
replica_jobs: int = 1
resume_completed: bool = False
class OCDocker.OCScore.Optimization.Protocol.ReplicatedProtocolResult(replica_results, summary_df, aggregate_summary, output_paths, failed_replicas=<factory>)[source]

Bases: object

Aggregated result returned by ReplicatedStagedProtocol.

Parameters:

  • replica_results (list[ReplicaResult]) – Per-replica result objects.

  • summary_df (pd.DataFrame) – One-row-per-replica summary table.

  • aggregate_summary (dict[str, Any]) – Mean/std metrics and separate best PDBbind/DUDEz replica selections.

  • output_paths (dict[str, str]) – Top-level report paths written by the replicated protocol.

  • failed_replicas (list[ReplicaResult]) – Failed replica result objects. Empty if all replicas succeeded.

replica_results: list[ReplicaResult]
summary_df: DataFrame
aggregate_summary: dict[str, Any]
output_paths: dict[str, str]
failed_replicas: list[ReplicaResult]
class OCDocker.OCScore.Optimization.Protocol.ReplicatedStagedProtocol(stages, n_replicas=1, base_seed=None, replica_name_prefix='replica', continue_on_replica_failure=False, write_reports=True, write_protocol_log=True, replica_jobs=1, resume_completed=False, config=None)[source]

Bases: object

Run a staged Optuna/modeling protocol across independent replicas.

Replica execution applies to the staged Optuna/modeling protocol only. The reduced datasets and selected features are reused from the input context; feature reduction is not repeated. Replica i receives seed base_seed + i and output directory Path(context.output_dir) / f"{replica_name_prefix}_{i:03d}".

Parameters:

  • stages (Iterable[ProtocolStage]) – Ordered stages for one full modeling protocol execution.

  • n_replicas (int, optional) – Number of independent protocol executions, by default 1.

  • base_seed (int | None, optional) – Base seed used for deterministic replica seeds. If None, use the input context random seed.

  • replica_name_prefix (str, optional) – Replica name prefix, by default "replica".

  • continue_on_replica_failure (bool, optional) – Continue after failed replicas and record the failures, by default False.

  • write_reports (bool, optional) – Write top-level replica summary reports, by default True.

  • write_protocol_log (bool, optional) – Write each replica protocol_log.json, by default True.

  • config (ReplicatedProtocolConfig | None, optional) – Optional configuration object. Explicit keyword arguments are used when config is None.

  • replica_jobs (int)

  • resume_completed (bool)

__init__(stages, n_replicas=1, base_seed=None, replica_name_prefix='replica', continue_on_replica_failure=False, write_reports=True, write_protocol_log=True, replica_jobs=1, resume_completed=False, config=None)[source]

Initialize a replicated staged protocol runner.

Parameters:

  • stages (Iterable[ProtocolStage]) – Ordered stages for one full modeling protocol execution.

  • n_replicas (int, optional) – Number of independent protocol executions, by default 1.

  • base_seed (int | None, optional) – Base seed for deterministic replica seeds, by default None.

  • replica_name_prefix (str, optional) – Replica name prefix, by default "replica".

  • continue_on_replica_failure (bool, optional) – Continue after failed replicas, by default False.

  • write_reports (bool, optional) – Write top-level replica reports, by default True.

  • write_protocol_log (bool, optional) – Write per-replica protocol logs, by default True.

  • replica_jobs (int, optional) – Number of replicas to execute concurrently, by default 1.

  • resume_completed (bool, optional) – Reuse completed replica outputs instead of rerunning them, by default False.

  • config (ReplicatedProtocolConfig | None, optional) – Optional config object, by default None.

Return type:

None

stages: list[ProtocolStage]
config: ReplicatedProtocolConfig
run(context)[source]

Run all replicas and aggregate their results.

Parameters:

context (ProtocolContext) – Base context containing reduced datasets, selected features, output directory, seed, and feature-reduction metadata.

Returns:

Per-replica contexts, summary table, aggregate metrics, output report paths, and failed replica records.

Return type:

ReplicatedProtocolResult

class OCDocker.OCScore.Optimization.Protocol.StagedProtocol(stages, write_protocol_log=True)[source]

Bases: object

Sequential protocol runner with explicit stage ownership.

Parameters:

  • stages (Iterable[ProtocolStage]) – Ordered protocol stages.

  • write_protocol_log (bool, optional) – If True, write protocol_log.json after the run, by default True.

stages: Iterable[ProtocolStage]
write_protocol_log: bool = True
run(context)[source]

Run all stages in order.

Parameters:

context (ProtocolContext) – Initial protocol context.

Returns:

Updated context after all stages have completed.

Return type:

ProtocolContext