Modeling source-specific characteristics for intelligent integration layer orchestration
Source Delivery Properties: When Is My Data Actually Ready?
Modeling source-specific characteristics for intelligent integration layer orchestration
The EOM Snapshot Problem
It’s the last working day of the month. Your finance team needs the end-of-month snapshot processed by 6 AM. But here’s the challenge:
- GL Transactions delivers daily, including weekends
- FX Rates only delivers on working days
- Positions delivers weekdays, but Friday’s file contains Thursday’s data
- Market Data is optional enrichment — nice to have, but don’t wait for it
Your integration pipeline has dependencies on all four sources. When can it actually start?
This is the source delivery properties problem. Every source system has unique delivery characteristics, and your orchestration needs to understand them.
Beyond Simple Dependencies
Modern orchestration tools handle dependencies well. “Wait for upstream_pipeline to succeed before starting downstream_pipeline.” Simple AND/OR logic.
But source data is more nuanced:
Source Delivery Characteristics:
┌─────────────────────────────────────────────────────────────────┐
│ │
│ GL_TRANSACTIONS FX_RATES POSITIONS │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Daily │ │ Weekdays │ │ Weekdays │ │
│ │ T+0 │ │ T+0 │ │ T+1 │ │
│ │ Mandatory │ │ Mandatory │ │ Mandatory │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
│ MARKET_DATA COUNTERPARTY_RATINGS │
│ ┌─────────────┐ ┌─────────────┐ │
│ │ Daily │ │ Weekly │ │
│ │ T+0 │ │ T+7 OK │ │
│ │ Optional │ │ Enrichment │ │
│ └─────────────┘ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
Each source has:
- Delivery schedule — When does data arrive?
- Reference date — What date does the data represent?
- Dependency class — Must we wait for it?
- Freshness tolerance — How stale can it be?
The Three Dependency Classes
Not all sources are created equal. We classify them:
Mandatory Sources
Definition : Processing cannot begin without this data.
source:
entity_id: ENT_gl_transactions
dependency_class: mandatory
delivery_schedule: daily
max_age_hours: 24
If GL transactions aren’t present for the snapshot period, we don’t start. Period.
Optional Sources
Definition : Process with available data, but don’t block if missing.
source:
entity_id: ENT_market_data
dependency_class: optional
delivery_schedule: daily
max_age_hours: 48
staleness_warning_hours: 12
Market data enriches our output, but yesterday’s prices are acceptable if today’s aren’t available.
Enrichment Sources
Definition : Nice to have, never blocks processing.
source:
entity_id: ENT_counterparty_ratings
dependency_class: enrichment
delivery_schedule: weekly
staleness_warning_hours: 168 # 1 week
Credit ratings update weekly. We use whatever’s available — even month-old data won’t block processing.
The Reference Date Problem
Here’s where it gets interesting. When we ask “is data complete for March 31st?”, we need to understand what each source means by “March 31st data.”
Calendar Day vs Working Day
March 2026:
Sun Mon Tue Wed Thu Fri Sat
29 30 31 1 2 3 4
▲
└── March 31st is a Tuesday
For sources delivering on working days:
- "Last working day of March" = March 31st ✓
But what if March 31st were a Sunday?
- Calendar day: Expect data for March 31st
- Last working day: Expect data for March 27th (Friday)
T+0 vs T+1 Delivery
Some sources deliver today’s data today (T+0). Others deliver yesterday’s data today (T+1).
# FX Rates: T+0 delivery
# File received March 31st contains March 31st rates
reference_date = snapshot_date # Same day
# Positions: T+1 delivery
# File received March 31st contains March 30th positions
reference_date = snapshot_date - timedelta(days=1)
For an EOM snapshot, the Positions source on March 31st contains March 30th data. We actually need to wait until April 1st to get March 31st positions.
Business Calendar Integration
Working day calculations require business calendars. Different regions have different holidays.
from dataclasses import dataclass
from datetime import date, timedelta
from typing import List, Set
@dataclass
class BusinessCalendar:
"""Business calendar with holidays."""
calendar_id: str
working_days: List[int] = None # 0=Mon, 4=Fri
holidays: Set[date] = None
def __post_init__(self):
self.working_days = self.working_days or [0, 1, 2, 3, 4]
self.holidays = self.holidays or set()
def is_working_day(self, d: date) -> bool:
if d.weekday() not in self.working_days:
return False
return d not in self.holidays
def last_working_day(self, d: date) -> date:
while not self.is_working_day(d):
d = d - timedelta(days=1)
return d
def last_working_day_of_month(self, year: int, month: int) -> date:
if month == 12:
last = date(year + 1, 1, 1) - timedelta(days=1)
else:
last = date(year, month + 1, 1) - timedelta(days=1)
return self.last_working_day(last)
Now we can calculate reference dates correctly:
# Netherlands calendar with public holidays
nl_calendar = BusinessCalendar(
calendar_id="NL",
holidays={
date(2026, 4, 27), # King's Day
date(2026, 5, 5), # Liberation Day
date(2026, 5, 14), # Ascension
# ... etc
}
)
# What's the last working day of April 2026?
# April 30th is Thursday, but April 27th is King's Day
april_lwd = nl_calendar.last_working_day_of_month(2026, 4)
# Result: April 30th (Thursday) - King's Day doesn't affect month end
The Readiness Check Pattern
Bring it all together with a readiness check:
@dataclass
class SourceReadinessResult:
"""Result of checking if source is ready."""
source_entity_id: str
snapshot_date: date
reference_date: date
status: str # NOT_EXPECTED, EXPECTED, COMPLETE, STALE
has_data: bool
data_age_hours: float | None = None
blocks_processing: bool = True
message: str | None = None
def check_source_readiness(
source_config: SourceConfig,
snapshot_date: date,
calendar: BusinessCalendar,
) -> SourceReadinessResult:
"""Check if source is ready for snapshot processing."""
# 1. Calculate reference date
if source_config.reference_type == "last_working_day":
ref_date = calendar.last_working_day(snapshot_date)
elif source_config.reference_type == "month_end_working":
ref_date = calendar.last_working_day_of_month(
snapshot_date.year, snapshot_date.month
)
else:
ref_date = snapshot_date - timedelta(source_config.offset_days)
# 2. Check if we expect data for this date
if source_config.delivery_schedule == "weekdays":
if not calendar.is_working_day(ref_date):
return SourceReadinessResult(
source_entity_id=source_config.entity_id,
snapshot_date=snapshot_date,
reference_date=ref_date,
status="NOT_EXPECTED",
has_data=False,
blocks_processing=False,
message=f"No delivery expected for {ref_date} (weekend/holiday)"
)
# 3. Check data presence and freshness
data_info = query_source_metadata(source_config.entity_id, ref_date)
if not data_info.has_data:
return SourceReadinessResult(
source_entity_id=source_config.entity_id,
snapshot_date=snapshot_date,
reference_date=ref_date,
status="EXPECTED",
has_data=False,
blocks_processing=source_config.dependency_class == "mandatory"
)
# 4. Check staleness
age_hours = (datetime.utcnow() - data_info.loaded_at).total_seconds() / 3600
is_stale = age_hours > source_config.max_age_hours
return SourceReadinessResult(
source_entity_id=source_config.entity_id,
snapshot_date=snapshot_date,
reference_date=ref_date,
status="STALE" if is_stale else "COMPLETE",
has_data=True,
data_age_hours=age_hours,
blocks_processing=is_stale and source_config.dependency_class == "mandatory"
)
Snapshot Readiness Report
For a complete picture, aggregate all sources:
┌─────────────────────────────────────────────────────────────────┐
│ EOM Snapshot Readiness: March 2026 │
│ Snapshot Date: 2026-03-31 │
│ Checked: 2026-03-31 05:30 UTC │
├─────────────────────────────────────────────────────────────────┤
│ Overall Status: BLOCKED │
│ Ready to Process: NO │
├─────────────────────────────────────────────────────────────────┤
│ MANDATORY SOURCES (2/3 ready) │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ [COMPLETE] GL_TRANSACTIONS │ │
│ │ Ref: 2026-03-31 | Age: 2.5h | Rows: 45,231 │ │
│ │ │ │
│ │ [COMPLETE] FX_RATES │ │
│ │ Ref: 2026-03-31 | Age: 4.1h | Rows: 892 │ │
│ │ │ │
│ │ [EXPECTED] POSITIONS ◄── BLOCKING │ │
│ │ Ref: 2026-03-30 | Waiting for T+1 delivery │ │
│ │ Expected by: 2026-03-31 08:00 UTC │ │
│ └──────────────────────────────────────────────────────────┘ │
├─────────────────────────────────────────────────────────────────┤
│ OPTIONAL SOURCES (1/1 ready) │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ [COMPLETE] MARKET_DATA │ │
│ │ Ref: 2026-03-31 | Age: 1.2h | Rows: 12,450 │ │
│ └──────────────────────────────────────────────────────────┘ │
├─────────────────────────────────────────────────────────────────┤
│ ENRICHMENT SOURCES (1/1 available) │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ [STALE] COUNTERPARTY_RATINGS │ │
│ │ Ref: 2026-03-24 | Age: 168h | Rows: 5,420 │ │
│ │ (Within tolerance - weekly update) │ │
│ └──────────────────────────────────────────────────────────┘ │
├─────────────────────────────────────────────────────────────────┤
│ Next Check: 2026-03-31 05:45 UTC (15 min interval) │
│ Estimated Ready: 2026-03-31 08:15 UTC │
└─────────────────────────────────────────────────────────────────┘
Trigger Integration
Connect readiness checks to your orchestration:
class SourceReadinessTrigger:
"""Trigger that fires when sources are ready."""
def create_trigger(
self,
pipeline_id: str,
source_ids: list[str],
snapshot_type: str = "EOM",
all_mandatory_required: bool = True,
poll_interval_minutes: int = 15,
max_wait_hours: int = 24,
) -> str:
"""Create source readiness trigger."""
return self.trigger_manager.create_custom_trigger(
name=f"{pipeline_id}_source_readiness",
pipeline_id=pipeline_id,
config={
"type": "source_readiness",
"sources": source_ids,
"snapshot_type": snapshot_type,
"all_mandatory": all_mandatory_required,
"poll_interval": poll_interval_minutes,
"max_wait": max_wait_hours,
}
)
def evaluate_trigger(
self,
trigger_id: str,
snapshot_date: date,
) -> bool:
"""Check if trigger should fire."""
config = self.get_trigger_config(trigger_id)
report = self.delivery_manager.check_snapshot_readiness(
snapshot_date=snapshot_date,
source_ids=config["sources"],
)
if config["all_mandatory"]:
return report.ready_to_process
else:
# Fire if any mandatory source is ready
return report.mandatory_ready > 0
YAML Configuration
Define source delivery properties in your metadata:
# sources/finance/gl_transactions.yaml
source:
entity_id: ENT_gl_transactions
name: General Ledger Transactions
delivery:
schedule: daily
reference_date_type: calendar_day
reference_date_offset: 0 # T+0
dependency:
classification: mandatory
max_age_hours: 24
staleness_warning_hours: 12
calendar:
calendar_id: NL
skip_holidays: false
# sources/finance/positions.yaml
source:
entity_id: ENT_positions
name: Trading Positions
delivery:
schedule: weekdays
delivery_time_utc: "08:00"
reference_date_type: calendar_day
reference_date_offset: 1 # T+1
dependency:
classification: mandatory
max_age_hours: 36 # Allow for weekend delays
calendar:
calendar_id: NL
skip_holidays: true
# sources/market/market_data.yaml
source:
entity_id: ENT_market_data
name: Market Data Feed
delivery:
schedule: daily
reference_date_type: last_working_day
dependency:
classification: optional
max_age_hours: 48
staleness_warning_hours: 12
fallback_source_id: ENT_market_data_backup
Pipeline Integration
Wire it into your pipeline definition:
# pipelines/eom_financial_snapshot.yaml
pipeline:
name: eom_financial_snapshot
description: End-of-month financial data consolidation
trigger:
type: source_readiness
snapshot_type: EOM
all_mandatory_required: true
poll_interval_minutes: 15
max_wait_hours: 48
sources:
mandatory:
- ENT_gl_transactions
- ENT_fx_rates
- ENT_positions
optional:
- ENT_market_data
enrichment:
- ENT_counterparty_ratings
on_ready:
# When all mandatory sources are ready
start_pipeline: true
notify:
- channel: "#data-platform"
message: "EOM snapshot starting - all sources ready"
on_timeout:
# If max_wait_hours exceeded
action: notify_and_wait
notify:
- channel: "#data-platform"
message: "EOM snapshot blocked - waiting for: {blocking_sources}"
Key Patterns
1. Weekend Handling
For weekday-only sources, don’t expect data on weekends:
if delivery_schedule == "weekdays":
if snapshot_date.weekday() >= 5: # Sat/Sun
# Reference date is Friday
reference_date = snapshot_date - timedelta(
days=snapshot_date.weekday() - 4
)
2. Month-End Working Day
For EOM snapshots that need working day data:
def last_working_day_of_month(year: int, month: int) -> date:
# Get last calendar day
if month == 12:
last = date(year + 1, 1, 1) - timedelta(days=1)
else:
last = date(year, month + 1, 1) - timedelta(days=1)
# Walk back to working day
while last.weekday() >= 5 or last in holidays:
last = last - timedelta(days=1)
return last
3. Fallback Sources
When primary source fails, use backup:
if primary_result.status == "FAILED":
if config.fallback_source_id:
fallback_result = check_source_readiness(
config.fallback_source_id,
snapshot_date
)
if fallback_result.has_data:
return fallback_result.with_warning(
"Using fallback source"
)
Benefits
- Accurate Timing : Start processing exactly when data is ready, not before
- Fewer Failures : Don’t attempt processing with incomplete data
- Clear Dependencies : Document what’s mandatory vs optional
- Audit Trail : Record which sources were available at snapshot time
- Intelligent Retries : Know when to check again vs when to alert
Conclusion
Source delivery properties bridge the gap between “pipeline dependency management” and “data readiness.” By modeling:
- Delivery schedules (daily, weekdays, weekly)
- Reference date logic (T+0, T+1, last working day)
- Dependency classes (mandatory, optional, enrichment)
- Business calendars (holidays, working days)
…you can answer the question every data team asks: “When is my data actually ready?”
The integration layer becomes smarter. It knows that Friday’s weekend source data won’t arrive until Monday. It understands that EOM snapshots need the last working day, not calendar month-end. It starts processing precisely when conditions are met.
That’s the power of metadata-driven orchestration.
Next in this series: “Snapshot Period Management: Beyond Simple Date Ranges”