Jobs

The jobs module provides async job management for long-running model fitting operations using ARQ (Async Redis Queue).

Job Management

Job management system for running MMM models in separate processes.

Provides: - Background model fitting with progress tracking - Multiple concurrent job management - Job persistence and recovery - Result storage and retrieval

class mmm_framework.jobs.JobStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

Status of a model fitting job.

PENDING = 'pending'

RUNNING = 'running'

COMPLETED = 'completed'

FAILED = 'failed'

CANCELLED = 'cancelled'

class mmm_framework.jobs.JobProgress(stage='initializing', current_step=0, total_steps=100, message='', started_at=None, updated_at=None)

Bases: object

Progress information for a running job.

stage: str = 'initializing'

current_step: int = 0

total_steps: int = 100

message: str = ''

started_at: str | None = None

updated_at: str | None = None

property percent_complete: float

to_dict()

Return type:

dict

classmethod from_dict(data)

Return type:

JobProgress

__init__(stage='initializing', current_step=0, total_steps=100, message='', started_at=None, updated_at=None)

class mmm_framework.jobs.JobConfig(data_path=None, n_chains=4, n_draws=1000, n_tune=1000, target_accept=0.95, use_numpyro=False, random_seed=42, trend_type='linear', trend_settings=<factory>, yearly_order=2, pool_geo=True, name='', description='', tags=<factory>)

Bases: object

Configuration for a model fitting job.

data_path: str | None = None

n_chains: int = 4

n_draws: int = 1000

n_tune: int = 1000

target_accept: float = 0.95

use_numpyro: bool = False

random_seed: int = 42

trend_type: str = 'linear'

trend_settings: dict

yearly_order: int = 2

pool_geo: bool = True

name: str = ''

description: str = ''

tags: list[str]

to_dict()

Return type:

dict

classmethod from_dict(data)

Return type:

JobConfig

__init__(data_path=None, n_chains=4, n_draws=1000, n_tune=1000, target_accept=0.95, use_numpyro=False, random_seed=42, trend_type='linear', trend_settings=<factory>, yearly_order=2, pool_geo=True, name='', description='', tags=<factory>)

class mmm_framework.jobs.JobResult(divergences=0, rhat_max=1.0, ess_bulk_min=0.0, r_squared=0.0, rmse=0.0, mape=0.0, fit_duration_seconds=0.0, trace_path=None, contributions_path=None, summary_path=None, error_message=None, error_traceback=None)

Bases: object

Results from a completed model fitting job.

divergences: int = 0

rhat_max: float = 1.0

ess_bulk_min: float = 0.0

r_squared: float = 0.0

rmse: float = 0.0

mape: float = 0.0

fit_duration_seconds: float = 0.0

trace_path: str | None = None

contributions_path: str | None = None

summary_path: str | None = None

error_message: str | None = None

error_traceback: str | None = None

to_dict()

Return type:

dict

classmethod from_dict(data)

Return type:

JobResult

__init__(divergences=0, rhat_max=1.0, ess_bulk_min=0.0, r_squared=0.0, rmse=0.0, mape=0.0, fit_duration_seconds=0.0, trace_path=None, contributions_path=None, summary_path=None, error_message=None, error_traceback=None)

class mmm_framework.jobs.Job(id, status, config, progress, result=None, created_at='', started_at=None, completed_at=None, pid=None)

Bases: object

Represents a model fitting job.

id: str

status: JobStatus

config: JobConfig

progress: JobProgress

result: JobResult | None = None

created_at: str = ''

started_at: str | None = None

completed_at: str | None = None

pid: int | None = None

property is_active: bool

property display_name: str

property duration_seconds: float | None

to_dict()

Return type:

dict

classmethod from_dict(data)

Return type:

Job

__init__(id, status, config, progress, result=None, created_at='', started_at=None, completed_at=None, pid=None)

class mmm_framework.jobs.JobManager(jobs_dir=None)

Bases: object

Manages multiple model fitting jobs.

Jobs are persisted to disk so they survive app restarts. Each job runs in a separate process.

__init__(jobs_dir=None)

create_job(panel, config)

Create a new job (but don’t start it yet).

Return type:

Job

Parameters

panelPanelDataset

The panel data to fit.

configJobConfig

Job configuration.

Returns

Job

The created job.

start_job(job_id)

Start a pending job.

Return type:

bool

Parameters

job_idstr

ID of the job to start.

Returns

bool

True if job was started successfully.

submit_job(panel, config)

Create and immediately start a job.

Return type:

Job

Parameters

panelPanelDataset

The panel data to fit.

configJobConfig

Job configuration.

Returns

Job

The created and started job.

get_job(job_id)

Get a job by ID.

Return type:

Job | None

Parameters

job_idstr

The job ID.

Returns

Job or None

The job, or None if not found.

list_jobs(status_filter=None, limit=None, order_by='created_at', ascending=False)

List all jobs.

Return type:

list[Job]

Parameters

status_filterlist[JobStatus], optional

Only return jobs with these statuses.

limitint, optional

Maximum number of jobs to return.

order_bystr

Field to sort by (created_at, started_at, completed_at).

ascendingbool

Sort order.

Returns

list[Job]

List of jobs.

cancel_job(job_id)

Cancel a running or pending job.

Return type:

bool

Parameters

job_idstr

The job ID.

Returns

bool

True if job was cancelled.

delete_job(job_id, force=False)

Delete a job and its files.

Return type:

bool

Parameters

job_idstr

The job ID.

forcebool

If True, delete even if running.

Returns

bool

True if job was deleted.

load_job_results(job_id)

Load full results for a completed job.

Return type:

dict | None

Parameters

job_idstr

The job ID.

Returns

dict or None

Dictionary with ‘mmm’, ‘results’, ‘panel’, ‘contributions’ keys, or None if not found or not completed.

get_active_jobs()

Get all currently active (pending or running) jobs.

Return type:

list[Job]

get_completed_jobs()

Get all completed jobs.

Return type:

list[Job]

cleanup_old_jobs(max_age_days=30, max_jobs=50)

Remove old completed/failed jobs.

Parameters

max_age_daysint

Remove jobs older than this many days.

max_jobsint

Keep at most this many completed jobs.

mmm_framework.jobs.get_job_manager(jobs_dir=None)

Get or create the default job manager.

Return type:

JobManager

mmm_framework.jobs.submit_model_job(panel, name='', description='', n_chains=4, n_draws=1000, n_tune=1000, target_accept=0.95, use_numpyro=False, trend_type='linear', trend_settings=None, yearly_order=2, pool_geo=True, random_seed=42, tags=None)

Convenience function to submit a model fitting job.

Return type:

Job

Parameters

panelPanelDataset

The panel data.

namestr

Job name.

descriptionstr

Job description.

n_chainsint

Number of MCMC chains.

n_drawsint

Number of draws per chain.

n_tuneint

Number of tuning samples.

target_acceptfloat

Target acceptance rate.

use_numpyrobool

Use NumPyro backend.

trend_typestr

Trend type (none, linear, piecewise, spline, gaussian_process).

trend_settingsdict

Trend-specific settings.

yearly_orderint

Fourier order for yearly seasonality.

pool_geobool

Enable hierarchical geo pooling.

random_seedint

Random seed.

tagslist[str]

Tags for the job.

Returns

Job

The submitted job.