Practical examples for running the Conflict Early Warning Pipeline.
- Basic Usage
- Phase Control
- Date Range Specification
- Diagnostic Modes
- Advanced Workflows
- Output Interpretation
- Command Reference
- Common Issues and Solutions
Run all phases for the default date range (configured in configs/data.yaml):
python main.pyExpected runtime: 2-4 hours
Output: Feature matrices, trained models, predictions in data/processed/
Test the pipeline with a small date range:
python main.py --start-date 2024-01-01 --end-date 2024-03-31Expected runtime: 20-30 minutes
The pipeline has 5 phases that can be selectively enabled/disabled:
# Skip static data ingestion (useful if already run)
python main.py --skip-static
# Skip dynamic data ingestion
python main.py --skip-dynamic
# Skip feature engineering
python main.py --skip-features
# Skip modeling and prediction
python main.py --skip-modeling
# Skip post-run analysis
python main.py --skip-analysis# Run only static ingestion
python main.py --skip-dynamic --skip-features --skip-modeling --skip-analysis
# Run only modeling (assumes feature matrix exists)
python main.py --skip-static --skip-dynamic --skip-features --skip-analysis
# Run everything except analysis (faster iteration)
python main.py --skip-analysis# Common: Re-run feature engineering and modeling only
python main.py \
--skip-static \
--skip-dynamic \
--skip-analysisThe pipeline uses an incremental-by-default strategy for all dynamic data sources to minimize API load and processing time.
- Default Behavior: For each source, the pipeline queries the database for the
MAX(date)and only fetches new records between that date and the requested--end-date. - Force Refresh (
--no-incremental): Use this flag to override the incremental logic and refetch the entire window from the configuredstart_date. - Overlap Sources: To ensure data integrity for sources with late-arriving or reprocessed data, the following sources include a mandatory 14-day safety overlap during incremental fetches:
- GEE Environmental features
- Dynamic World land cover
- IOM DTM displacement
- GDELT (7-day buffer)
# Specify custom date window
python main.py \
--start-date 2020-01-01 \
--end-date 2023-12-31# Last 2 years of data
python main.py \
--start-date 2023-01-01 \
--end-date 2024-12-31# Pre-COVID period
python main.py \
--start-date 2015-01-01 \
--end-date 2019-12-31The pipeline includes specialized diagnostic modes for analyzing feature quality before model training. These modes automatically exclude structural break flags (data availability indicators) that add noise to VIF/correlation analysis.
Important: Diagnostic modes require an existing feature matrix. You must run the pipeline first:
# Option 1: Full pipeline
python main.py
# Option 2: Build matrix without modeling (faster)
python main.py --skip-modeling --skip-analysis# Run all diagnostics (VIF, correlation, stability analysis)
# Structural break flags excluded by default
python main.py --run-diagnostics-onlyOutput files in analysis/diagnostics/:
pairwise_dependence.csv- Feature pairs with |ρ| ≥ 0.6theme_redundancy.csv- Per-theme VIF, condition number, PCA variancetemporal_stability.csv- Correlation stability across time periodsinterpretability_checks.csv- Per-feature variance, entropy, max correlationexcluded_columns.csv- Columns excluded and reasons
# Just VIF and correlation analysis
python main.py --run-collinearity-onlyOutput files in data/processed/:
vif_analysis.csv- Full VIF analysiscorrelation_matrix.parquet- Complete correlation matrixfeature_clusters.json- Hierarchical clustering of correlated featuressuggested_removals.json- Features flagged for potential removal
# Include structural break flags (for comparison)
python main.py --run-diagnostics-only --include-structural-breaks
# Exclude additional columns beyond structural breaks
python main.py --run-diagnostics-only --diagnostic-exclude-cols "cw_score_local,gdelt_goldstein_mean"
# Use sampling for faster analysis on large datasets
python main.py --run-diagnostics-only --diagnostic-sample-frac 0.3# Show all configured structural break flags and their coverage
python main.py --list-structural-breaksExample output:
============================================================
STRUCTURAL BREAK FLAGS
============================================================
Total configured flags: 10
- econ_data_available
- food_data_available
- gdelt_data_available
- ioda_data_available
- iom_data_available
- is_viirs_available
- is_worldpop_v1
- landcover_avail
------------------------------------------------------------
Checking feature matrix for structural break flags...
Present in matrix: 7
✓ econ_data_available
✓ ioda_data_available
✓ iom_data_available
...
Coverage summary:
econ_data_available: 85.2% coverage (values: [0, 1])
iom_data_available: 72.3% coverage (values: [0, 1])
...
# Run feature diagnostics directly
python -m scripts.diagnostics.feature_diagnostics --exclude-structural-breaks
# Run collinearity check with custom thresholds
python -m scripts.collinearity_check \
--vif-threshold 5.0 \
--corr-threshold 0.8 \
--exclude-structural-breaks
# Specify custom parquet file
python -m scripts.diagnostics.feature_diagnostics \
--parquet data/processed/custom_matrix.parquetfrom pipeline.common.diagnostic_utils import (
filter_diagnostic_columns,
load_feature_matrix_for_diagnostics,
get_structural_break_flags,
summarize_structural_flags
)
# Load pre-filtered matrix
df = load_feature_matrix_for_diagnostics(
exclude_structural_breaks=True,
extra_exclusions=["col1", "col2"],
sample_frac=0.5
)
# Or filter an existing DataFrame
df_filtered = filter_diagnostic_columns(
df,
exclude_structural_breaks=True,
extra_exclusions=["col1"]
)
# List configured flags
flags = get_structural_break_flags()
print(f"Structural break flags: {flags}")
# Get coverage summary
summary = summarize_structural_flags(df)
print(summary)Structural break flags are configured in configs/features.yaml:
diagnostics:
structural_break_flags:
- is_worldpop_v1
- iom_data_available
- econ_data_available
- ioda_data_available
- landcover_avail
- is_viirs_available
- gdelt_data_available
- food_data_available
exclude_from_diagnostics: [] # Add columns here
thresholds:
vif_warning: 5.0
vif_severe: 10.0
correlation_warning: 0.7
correlation_severe: 0.9Structural break flags are binary (0/1) columns indicating data availability periods:
| Flag | Purpose |
|---|---|
is_worldpop_v1 |
WorldPop V1 vs V2 methodology (pre-2015) |
iom_data_available |
IOM DTM coverage start (2018-01-31) |
econ_data_available |
Yahoo Finance coverage (2003-12-01) |
ioda_data_available |
IODA monitoring start (2022-01-01) |
landcover_data_available |
Dynamic World start (2017-04-01) |
viirs_data_available |
VIIRS coverage start (2012-01-28) |
These flags are essential for XGBoost training (teaching the model when data becomes available) but:
- Inflate VIF artificially - Binary flags with high prevalence correlate spuriously with many features
- Pollute correlation matrices - Show up as "highly correlated" with unrelated features
- Distort clustering - Create artificial feature clusters based on data availability
By default, diagnostic modes exclude them to give you a cleaner picture of actual feature relationships. Use --include-structural-breaks to include them when needed.
The dashboard now supports multi-level administrative aggregation, allowing users to move beyond raw H3 hexagons to scientifically defensible regional summaries.
In both the Predictions and Temporal Features tabs, use the View Level dropdown to switch between:
- H3 Hexagons: Raw 10km grid resolution.
- Regions: Legacy World Bank 17-unit aggregation.
- Prefectures: 2021 OCHA 20-unit aggregation.
- Subprefectures: 2021 OCHA 72-unit aggregation.
When an administrative level is selected, the system automatically applies the appropriate statistical method:
- Mean: For continuous risk scores and predicted probabilities.
- Median: For environmental anomalies (e.g., NDVI, Rainfall) to mitigate sensor noise.
- Sum: For additive counts (e.g., Fatalities, Population, Displacement).
In the Temporal Features tab, selecting a Subprefecture will automatically trigger a Hierarchical Comparison. The trend chart will overlay the subprefecture's data against its parent Prefecture's trend, allowing you to see if a local area is deviating from its regional norm.
The "Normalize" toggle uses intelligent scaling based on the data type:
- Positive Data (Prices, Counts): Uses Min-Max Scaling [0, 1] to show relative growth.
- Anomaly Data (Negative Values): Uses Max-Abs Scaling [-1, 1] to preserve the zero-line, ensuring that droughts or negative shocks correctly appear below zero.
You have existing models but want to refresh with latest data:
# Step 1: Ingest new data
python main.py \
--start-date 2025-01-01 \
--end-date 2025-12-04 \
--skip-modeling \
--skip-analysis
# Step 2: Generate predictions with existing models
python pipeline/modeling/generate_predictions.py 14d
python pipeline/modeling/generate_predictions.py 1m
python pipeline/modeling/generate_predictions.py 3m# Step 1: Edit configs/features.yaml to enable/disable features
nano configs/features.yaml
# Step 2: Rebuild feature matrix
python main.py \
--skip-static \
--skip-dynamic
# Step 3: Verify features
python -c "import pandas as pd; df = pd.read_parquet('data/processed/feature_matrix.parquet'); print(df.columns.tolist())"# Step 1: Edit configs/models.yaml
nano configs/models.yaml
# Step 2: Retrain models only
python main.py \
--skip-static \
--skip-dynamic \
--skip-features# Step 1: Reset database schema
python main.py --reset-schema
# Step 2: Run full pipeline
python main.py--reset-schema deletes ALL data in the car_cewp schema.
# Check prerequisites without running pipeline
python main.py --validate-only
# Example output:
# ✅ Validation passed for planned execution path.
# OR
# ❌ Validation failed. Missing table: car_cewp.features_staticLocation: data/processed/feature_matrix.parquet
# Inspect structure
python -c "
import pandas as pd
df = pd.read_parquet('data/processed/feature_matrix.parquet')
print(f'Shape: {df.shape}')
print(f'Date range: {df.date.min()} to {df.date.max()}')
print(f'H3 cells: {df.h3_index.nunique()}')
print(f'Features: {len(df.columns) - 2}') # Minus h3_index and date
"Locations:
data/processed/predictions_14d.parquet- 14-day forecastdata/processed/predictions_1m.parquet- 1-month forecastdata/processed/predictions_3m.parquet- 3-month forecast
Each forecast includes:
| Column | Type | Description | Example |
|---|---|---|---|
h3_index |
BIGINT | Hexagonal cell identifier | 599686042433355775 |
date |
DATE | Forecast target date | 2024-06-15 |
onset_prob |
FLOAT | Calibrated conflict probability | 0.0342 |
predicted_fatalities |
FLOAT | Expected fatality count (capped at 500) | 2.47 |
fatalities_lower |
FLOAT | BCCP 90% interval lower bound | 0.0 |
fatalities_upper |
FLOAT | BCCP 90% interval upper bound | 12.3 |
risk_tier |
TEXT | Operational priority tier | Critical, High, Elevated, Low |
is_priority_target |
BOOLEAN | Binary flag for Critical tier (Top 5%) | true / false |
The system uses adaptive quantile-based tiering to translate probabilities into actionable priorities:
| Tier | Coverage | Description | Operational Response |
|---|---|---|---|
| Critical | Top 5% | Highest-risk cells requiring immediate attention | Field verification within 24-48 hours; immediate JMAC escalation |
| High | Top 15% | Elevated risk warranting heightened monitoring | Daily monitoring; preemptive stakeholder engagement; logistics prep |
| Elevated | Top 30% | Emerging risk trends for watchlisting | Weekly situation reports; pattern analysis; open-source monitoring |
| Low | Bottom 70% | Baseline monitoring | Standard surveillance |
Key Feature: Tiers adapt to the country's violence trajectory. During escalation, Critical tier cells may have onset_prob > 0.45; during calm periods, onset_prob > 0.28 may qualify. Both represent the top 5% of relative risk, preventing alert fatigue while maintaining sensitivity.
# View Critical tier cells (highest priority)
python -c "
import pandas as pd
df = pd.read_parquet('data/processed/predictions_14d.parquet')
critical = df[df['risk_tier'] == 'Critical'].sort_values('onset_prob', ascending=False)
print(critical[['h3_index', 'date', 'onset_prob', 'predicted_fatalities', 'risk_tier']].head(10))
"
# View tier distribution
python -c "
import pandas as pd
df = pd.read_parquet('data/processed/predictions_14d.parquet')
print(df['risk_tier'].value_counts(normalize=True).round(3))
"
# Filter to priority targets only
python -c "
import pandas as pd
df = pd.read_parquet('data/processed/predictions_14d.parquet')
priority = df[df['is_priority_target'] == True]
print(f'Priority cells: {len(priority):,} (~{len(priority)/len(df)*100:.1f}%)')
"# Convert to GeoJSON for QGIS/ArcGIS
python -c "
import pandas as pd
import geopandas as gpd
import h3.api.basic_int as h3
from shapely.geometry import Polygon
# Load predictions
df = pd.read_parquet('data/processed/predictions_14d.parquet')
# Filter to priority tiers
df = df[df['risk_tier'].isin(['Critical', 'High', 'Elevated'])]
# Convert H3 to polygons
geometries = []
for h3_idx in df['h3_index']:
boundary = h3.h3_to_geo_boundary(h3_idx, geo_json=True)
geometries.append(Polygon(boundary))
gdf = gpd.GeoDataFrame(df, geometry=geometries, crs='EPSG:4326')
gdf.to_file('predictions_priority_tiers.geojson', driver='GeoJSON')
print('Exported to predictions_priority_tiers.geojson')
"Location: models/
ls -lh models/
# two_stage_ensemble_14d_xgboost.pkl
# two_stage_ensemble_1m_xgboost.pkl
# two_stage_ensemble_3m_xgboost.pklLocation: analysis/
Generated plots include:
macro_feature_importance.png- Top-level feature importancesubtheme_shap_*.png- SHAP values by feature categorymodel_comparison_*.png- Model performance metricsspatial_residuals_*.png- Geographic error patterns
python main.py --help
Options:
--start-date TEXT Override start date (YYYY-MM-DD)
--end-date TEXT Override end date (YYYY-MM-DD)
--reset-schema Drop and recreate database schema
--skip-static Skip static data ingestion
--skip-dynamic Skip dynamic data ingestion
--skip-gdelt-themes Skip GDELT themes fetch
--skip-features Skip feature engineering
--skip-modeling Skip model training and predictions
--skip-analysis Skip post-run analysis
--stop-after-features Run feature engineering, write schema summary, exit
--stop-after-feature-matrix Build feature matrix, then exit before modeling
--step [acled_hybrid|build_matrix]
Run a single pipeline step and exit
--validate-only Check prerequisites without running
# Diagnostic Options
--run-diagnostics-only Run VIF/correlation diagnostics and exit
--run-collinearity-only Run collinearity check only and exit
--exclude-structural-breaks Exclude data availability flags from diagnostics (default)
--include-structural-breaks Include structural break flags in diagnostics
--diagnostic-exclude-cols TEXT Comma-separated columns to exclude from diagnostics
--diagnostic-sample-frac FLOAT Sample fraction for diagnostics (0.0-1.0)
--list-structural-breaks List all structural break flags and exit
--help Show this message and exitMost pipeline components can be run standalone:
# Database initialization
python init_db.py
# Individual ingestion scripts
python pipeline/ingestion/fetch_acled.py
python pipeline/ingestion/fetch_gee_server_side.py
# Feature engineering
python pipeline/processing/process_spine_and_infrastructure.py
# Build feature matrix
python pipeline/modeling/build_feature_matrix.py
# Train models
python pipeline/modeling/train_models.py
# Generate predictions
python pipeline/modeling/generate_predictions.py 14d
# Run analysis
python pipeline/analysis/analyze_feature_importance.pyCause: You skipped too many phases or are running diagnostics without building the matrix first.
Solution:
# Build the feature matrix
python main.py --skip-modeling --skip-analysis
# Or rebuild just the matrix from existing DB
python main.py --step build_matrixCause: Diagnostic modes require an existing feature matrix. They don't build one from scratch.
Solution:
# Step 1: Build the pipeline first
python main.py --skip-modeling --skip-analysis
# Step 2: Then run diagnostics
python main.py --run-diagnostics-onlyCause: The database tables haven't been populated yet.
Solution: Run the full pipeline from the beginning:
python main.pySolution: You need to train models first:
python main.py --skip-static --skip-dynamic --skip-featuresSolution: Reduce date range to lower memory usage:
python main.py --start-date 2023-01-01 --end-date 2024-01-01Solution: GEE has daily limits. Wait 24 hours or use service account for higher quota.
Cause: Structural break flags may be included in the analysis.
Solution: Ensure you're excluding them (default behavior):
python main.py --run-diagnostics-only
# Or explicitly exclude
python main.py --run-collinearity-only --exclude-structural-breaksSolution: Use sampling for faster iteration:
python main.py --run-diagnostics-only --diagnostic-sample-frac 0.3# Test changes quickly with small date range
python main.py --start-date 2024-11-01 --end-date 2024-11-30 --skip-analysis# Watch logs in real-time
tail -f logs/pipeline.log# Check what's been ingested
psql -d car_cewp -c "SELECT COUNT(*) FROM car_cewp.acled_events;"
psql -d car_cewp -c "SELECT MIN(date), MAX(date) FROM car_cewp.temporal_features;"# Check YAML syntax
python -c "import yaml; yaml.safe_load(open('configs/data.yaml'))"