# vickyguo0907/ml_pipeline architecture

> Architecture diagram of the GitHub repository vickyguo0907/ml_pipeline, generated by GitDiagram from its file tree, README and a sample of source files (last updated 2026-11-11). It is an AI-made overview: check details against the source.

- Interactive diagram: https://gitdiagram.com/vickyguo0907/ml_pipeline
- Repository: https://github.com/vickyguo0907/ml_pipeline

## Overview

This is a config-driven ML pipeline template, with Airflow orchestrating data from landing files through validation, preparation, model training, evaluation, and registry staging; FastAPI serves promoted models. The main path is shared by tabular pipelines, with finance and forecasting implementations for distinct problem domains. Profiling, unsupervised exploration, drift reports, and benchmark comparisons are additional capabilities. The README documents the product workflow; the source index gives some module relationships, but excerpts are incomplete, so unsupported internal calls are omitted.

Main entry points: [`src/dags/dag_factory.py`](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/dags/dag_factory.py) (Airflow DAG factory), [`src/clean.py`](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/clean.py) (Tabular cleaning), [`src/features.py`](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/features.py) (Tabular features), [`src/train.py`](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/train.py) (Tabular training)

## Diagram (Mermaid)

```mermaid
flowchart TD

subgraph group_orchestration["Orchestration"]
  node_airflow["Airflow DAG factory<br/>[dag_factory.py]"]
end

subgraph group_shared["Shared ML pipeline"]
  node_ingest["Ingestion<br/>[ingest.py]"]
  node_rawcheck["Raw validation<br/>[validate.py]"]
  node_clean["Tabular cleaning<br/>[clean.py]"]
  node_features["Tabular features<br/>[features.py]"]
  node_featurecheck["Feature validation<br/>[features.py]"]
  node_training["Tabular training<br/>[train.py]"]
  node_evaluation["Tabular evaluation<br/>[evaluate.py]"]
  node_profiling["Data profiling<br/>[profile.py]"]
  node_exploration["Unsupervised exploration<br/>[explore.py]"]
  node_monitoring["Drift reporting<br/>[monitoring.py]"]
end

subgraph group_finance["Finance pipelines"]
  node_financeclean["Finance cleaning<br/>[clean_finance.py]"]
  node_financefeatures["Finance features"]
  node_financetraining["Finance training<br/>[train_finance.py]"]
  node_financeevaluation["Finance evaluation"]
  node_rankic["Rank IC<br/>[rank_ic.py]"]
end

subgraph group_forecast["Forecasting pipelines"]
  node_forecastclean["Forecast cleaning<br/>[clean_forecast.py]"]
  node_forecastfeatures["Forecast features"]
  node_weather["Weather processing<br/>[weather.py]"]
  node_forecasttraining["Forecast training<br/>[train_forecast.py]"]
  node_rolling["Rolling-origin evaluation<br/>[rolling_origin.py]"]
  node_forecastmodels["Forecast models<br/>[direct_forecast.py]"]
end

subgraph group_runtime["Serving and support"]
  node_registry[("MLflow registry<br/>[model_registry.py]")]
  node_serving["Prediction API<br/>[serve.py]"]
  node_config["Configuration models<br/>[config.py]"]
  node_storage[("Run artifacts and manifests<br/>[io.py]")]
  node_transforms["Shared transforms<br/>[transforms.py]"]
  node_diagnostics["Model diagnostics<br/>[diagnostics.py]"]
end

node_user(("Pipeline operator"))
node_landing["Landing data"]
node_operator["Predictions and reports"]

node_user -->|"triggers run"| node_airflow
node_landing -->|"stages files"| node_ingest
node_airflow -->|"orchestrates"| node_ingest
node_ingest -->|"validates raw"| node_rawcheck
node_airflow -->|"orchestrates"| node_clean
node_airflow -->|"orchestrates"| node_features
node_airflow -->|"validates features"| node_featurecheck
node_airflow -->|"orchestrates"| node_training
node_airflow -->|"orchestrates"| node_evaluation
node_training -->|"logs models"| node_registry
node_evaluation -->|"registers models"| node_registry
node_airflow -.->|"optionally runs"| node_profiling
node_airflow -.->|"optionally runs"| node_exploration
node_airflow -.->|"optionally reports drift"| node_monitoring
node_airflow -->|"dispatches finance"| node_financeclean
node_airflow -->|"dispatches finance"| node_financefeatures
node_airflow -->|"dispatches finance"| node_financetraining
node_airflow -->|"dispatches finance"| node_financeevaluation
node_financeevaluation -->|"computes rank IC"| node_rankic
node_airflow -->|"dispatches forecast"| node_forecastclean
node_airflow -->|"dispatches forecast"| node_forecastfeatures
node_airflow -->|"dispatches forecast"| node_forecasttraining
node_forecasttraining -->|"evaluates origins"| node_rolling
node_rolling -->|"uses forecast logic"| node_forecastmodels
node_forecastclean -->|"aggregates weather"| node_weather
node_forecastfeatures -->|"builds weather features"| node_weather
node_user -->|"requests prediction"| node_serving
node_serving -.->|"loads promoted model"| node_registry
node_serving -->|"returns prediction"| node_operator
node_clean -->|"reads and writes"| node_storage
node_features -->|"reads and writes"| node_storage
node_clean -->|"loads settings"| node_config
node_features -->|"applies transforms"| node_transforms
node_training -->|"uses diagnostics"| node_diagnostics

click node_airflow "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/dags/dag_factory.py"
click node_ingest "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/ingest.py"
click node_rawcheck "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/validate.py"
click node_clean "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/clean.py"
click node_features "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/features.py"
click node_featurecheck "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/schemas/features.py"
click node_training "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/train.py"
click node_evaluation "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/evaluate.py"
click node_registry "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/utils/model_registry.py"
click node_profiling "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/profile.py"
click node_exploration "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/explore.py"
click node_monitoring "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/monitoring.py"
click node_financeclean "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/finance/clean_finance.py"
click node_financefeatures "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/finance/features_finance.py"
click node_financetraining "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/finance/train_finance.py"
click node_financeevaluation "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/finance/evaluate_finance.py"
click node_rankic "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/finance/rank_ic.py"
click node_forecastclean "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/forecasting/clean_forecast.py"
click node_forecastfeatures "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/forecasting/features_forecast.py"
click node_weather "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/forecasting/weather.py"
click node_forecasttraining "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/forecasting/train_forecast.py"
click node_rolling "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/forecasting/rolling_origin.py"
click node_forecastmodels "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/forecasting/direct_forecast.py"
click node_serving "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/serve.py"
click node_config "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/utils/config.py"
click node_storage "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/utils/io.py"
click node_transforms "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/utils/transforms.py"
click node_diagnostics "https://github.com/vickyguo0907/ml_pipeline/blob/main/src/utils/diagnostics.py"

classDef toneNeutral fill:#f8fafc,stroke:#334155,stroke-width:1.5px,color:#0f172a
classDef toneBlue fill:#dbeafe,stroke:#2563eb,stroke-width:1.5px,color:#172554
classDef toneAmber fill:#fef3c7,stroke:#d97706,stroke-width:1.5px,color:#78350f
classDef toneMint fill:#dcfce7,stroke:#16a34a,stroke-width:1.5px,color:#14532d
classDef toneRose fill:#ffe4e6,stroke:#e11d48,stroke-width:1.5px,color:#881337
classDef toneIndigo fill:#e0e7ff,stroke:#4f46e5,stroke-width:1.5px,color:#312e81
classDef toneTeal fill:#ccfbf1,stroke:#0f766e,stroke-width:1.5px,color:#134e4a
class node_airflow toneBlue
class node_ingest,node_rawcheck,node_clean,node_features,node_featurecheck,node_training,node_evaluation,node_profiling,node_exploration,node_monitoring toneAmber
class node_financeclean,node_financefeatures,node_financetraining,node_financeevaluation,node_rankic toneMint
class node_forecastclean,node_forecastfeatures,node_weather,node_forecasttraining,node_rolling,node_forecastmodels toneRose
class node_registry,node_serving,node_config,node_storage,node_transforms,node_diagnostics,node_user,node_landing,node_operator toneIndigo
```

## Components

### Orchestration

- [Airflow DAG factory](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/dags/dag_factory.py): `src/dags/dag_factory.py`

### Shared ML pipeline

- [Ingestion](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/ingest.py): `src/ingest.py`
- [Raw validation](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/validate.py): `src/validate.py`
- [Tabular cleaning](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/clean.py): `src/clean.py`
- [Tabular features](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/features.py): `src/features.py`
- [Feature validation](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/schemas/features.py): `src/schemas/features.py`
- [Tabular training](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/train.py): `src/train.py`
- [Tabular evaluation](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/evaluate.py): `src/evaluate.py`
- [Data profiling](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/profile.py): `src/profile.py`
- [Unsupervised exploration](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/explore.py): `src/explore.py`
- [Drift reporting](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/monitoring.py): `src/monitoring.py`

### Finance pipelines

- [Finance cleaning](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/finance/clean_finance.py): `src/finance/clean_finance.py`
- [Finance features](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/finance/features_finance.py): `src/finance/features_finance.py`
- [Finance training](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/finance/train_finance.py): `src/finance/train_finance.py`
- [Finance evaluation](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/finance/evaluate_finance.py): `src/finance/evaluate_finance.py`
- [Rank IC](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/finance/rank_ic.py): `src/finance/rank_ic.py`

### Forecasting pipelines

- [Forecast cleaning](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/forecasting/clean_forecast.py): `src/forecasting/clean_forecast.py`
- [Forecast features](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/forecasting/features_forecast.py): `src/forecasting/features_forecast.py`
- [Weather processing](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/forecasting/weather.py): `src/forecasting/weather.py`
- [Forecast training](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/forecasting/train_forecast.py): `src/forecasting/train_forecast.py`
- [Rolling-origin evaluation](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/forecasting/rolling_origin.py): `src/forecasting/rolling_origin.py`
- [Forecast models](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/forecasting/direct_forecast.py): `src/forecasting/direct_forecast.py`

### Serving and support

- [MLflow registry](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/utils/model_registry.py): `src/utils/model_registry.py`
- [Prediction API](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/serve.py): `src/serve.py`
- [Configuration models](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/utils/config.py): `src/utils/config.py`
- [Run artifacts and manifests](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/utils/io.py): `src/utils/io.py`
- [Shared transforms](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/utils/transforms.py): `src/utils/transforms.py`
- [Model diagnostics](https://github.com/vickyguo0907/ml_pipeline/blob/main/src/utils/diagnostics.py): `src/utils/diagnostics.py`

### Other components

- **Pipeline operator**
- **Landing data**
- **Predictions and reports**

## Connections

- Pipeline operator → Airflow DAG factory: triggers run
- Landing data → Ingestion: stages files
- Airflow DAG factory → Ingestion: orchestrates (evidence: `src/dags/dag_factory.py`)
- Ingestion → Raw validation: validates raw
- Airflow DAG factory → Tabular cleaning: orchestrates (evidence: `src/dags/dag_factory.py`)
- Airflow DAG factory → Tabular features: orchestrates (evidence: `src/dags/dag_factory.py`)
- Airflow DAG factory → Feature validation: validates features (evidence: `src/dags/dag_factory.py`)
- Airflow DAG factory → Tabular training: orchestrates (evidence: `src/dags/dag_factory.py`)
- Airflow DAG factory → Tabular evaluation: orchestrates (evidence: `src/dags/dag_factory.py`)
- Tabular training → MLflow registry: logs models (evidence: `src/train.py`)
- Tabular evaluation → MLflow registry: registers models (evidence: `src/evaluate.py`)
- Airflow DAG factory → Data profiling: optionally runs (evidence: `src/dags/dag_factory.py`)
- Airflow DAG factory → Unsupervised exploration: optionally runs (evidence: `src/dags/dag_factory.py`)
- Airflow DAG factory → Drift reporting: optionally reports drift (evidence: `src/dags/dag_factory.py`)
- Airflow DAG factory → Finance cleaning: dispatches finance (evidence: `src/dags/dag_factory.py`)
- Airflow DAG factory → Finance features: dispatches finance (evidence: `src/dags/dag_factory.py`)
- Airflow DAG factory → Finance training: dispatches finance (evidence: `src/dags/dag_factory.py`)
- Airflow DAG factory → Finance evaluation: dispatches finance (evidence: `src/dags/dag_factory.py`)
- Finance evaluation → Rank IC: computes rank IC (evidence: `src/finance/evaluate_finance.py`)
- Airflow DAG factory → Forecast cleaning: dispatches forecast (evidence: `src/dags/dag_factory.py`)
- Airflow DAG factory → Forecast features: dispatches forecast (evidence: `src/dags/dag_factory.py`)
- Airflow DAG factory → Forecast training: dispatches forecast (evidence: `src/dags/dag_factory.py`)
- Forecast training → Rolling-origin evaluation: evaluates origins (evidence: `src/forecasting/train_forecast.py`)
- Rolling-origin evaluation → Forecast models: uses forecast logic (evidence: `src/forecasting/rolling_origin.py`)
- Forecast cleaning → Weather processing: aggregates weather (evidence: `src/forecasting/clean_forecast.py`)
- Forecast features → Weather processing: builds weather features (evidence: `src/forecasting/features_forecast.py`)
- Pipeline operator → Prediction API: requests prediction
- Prediction API → MLflow registry: loads promoted model
- Prediction API → Predictions and reports: returns prediction
- Tabular cleaning → Run artifacts and manifests: reads and writes (evidence: `src/clean.py`)
- Tabular features → Run artifacts and manifests: reads and writes (evidence: `src/features.py`)
- Tabular cleaning → Configuration models: loads settings (evidence: `src/clean.py`)
- Tabular features → Shared transforms: applies transforms (evidence: `src/features.py`)
- Tabular training → Model diagnostics: uses diagnostics (evidence: `src/train.py`)