Middleware: Seasonality Adjuster

Applies 52-week seasonality patterns to raw demand data. First step in the forecast pipeline.

Config [mw-seasonality-config]

# Input: expects 'data' array from previous pipeline step
# Each record should have: item_id, location_id, week, quantity, category
patterns:
  audio:
    - 0.8, 0.75, 0.7, 0.75, 0.8, 0.85, 0.9, 0.9
    - 0.95, 1.0, 1.0, 1.0, 1.0, 0.95, 0.9, 0.9
    - 0.9, 0.95, 1.0, 1.0, 0.95, 0.9, 0.9, 0.95
    - 1.0, 1.1, 1.2, 1.3, 1.2, 1.1, 1.0, 1.0
    - 1.0, 1.1, 1.1, 1.2, 1.2, 1.3, 1.4, 1.5
    - 1.7, 2.0, 2.2, 2.5, 2.8, 2.5, 2.0, 1.5
    - 1.2, 1.0, 0.9, 0.85
  accessories:
    - 0.85, 0.8, 0.75, 0.8, 0.85, 0.9, 0.95, 0.95
    - 1.0, 1.0, 1.0, 1.0, 1.0, 0.95, 0.95, 0.95
    - 0.95, 0.95, 1.0, 1.0, 0.95, 0.95, 0.95, 1.0
    - 1.0, 1.05, 1.1, 1.15, 1.15, 1.1, 1.05, 1.05
    - 1.1, 1.15, 1.2, 1.3, 1.3, 1.35, 1.4, 1.5
    - 1.7, 2.0, 2.3, 2.6, 2.8, 2.5, 2.0, 1.5
    - 1.2, 1.0, 0.9, 0.85
  default:
    - 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0
    - 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0
    - 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0
    - 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0
    - 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0
    - 1.2, 1.4, 1.6, 1.8, 2.0, 1.8, 1.5, 1.2
    - 1.0, 1.0, 1.0, 1.0
ttl_seconds: 3600

Fetch [mw-seasonality-fetch]

import json
import re
from datetime import datetime

# Config is unified: node config + pipeline params + session context
# PipelineProvider loads and merges these automatically
patterns = config.get('patterns', {})
default_pattern = patterns.get('default', [1.0] * 52)

# Input data comes from previous pipeline step (accumulated context)
# Each record should have: item_id, location_id, week, quantity, category
input_data = config.get('data', [])
if not input_data:
    raise ValueError("No input data - mw-seasonality requires 'data' from previous pipeline step")

raw_data = input_data

# Apply seasonality adjustment
adjusted = []
for record in raw_data:
    week_match = re.search(r'W(\d+)', record.get('week', 'W01'))
    week_num = int(week_match.group(1)) if week_match else 1
    week_idx = (week_num - 1) % 52

    # Category comes directly from the record (set by source)
    category = record.get('category', 'default')
    
    pattern = patterns.get(category, default_pattern)
    flat_pattern = []
    for p in pattern:
        if isinstance(p, list):
            flat_pattern.extend(p)
        elif isinstance(p, str):
            flat_pattern.extend([float(x.strip()) for x in p.split(',')])
        else:
            flat_pattern.append(float(p))
    
    factor = flat_pattern[week_idx] if week_idx < len(flat_pattern) else 1.0
    
    adj_record = dict(record)
    adj_record['raw_quantity'] = record.get('quantity', 0)
    adj_record['seasonality_factor'] = factor
    adj_record['adjusted_quantity'] = round(record.get('quantity', 0) / factor)
    adj_record['category'] = category
    adjusted.append(adj_record)

result = {
    'data': adjusted,
    'meta': {
        'generated_at': datetime.utcnow().isoformat(),
        'stage': 'seasonality',
        'input_count': len(raw_data),
        'output_count': len(adjusted),
        'patterns_applied': list(set(r['category'] for r in adjusted)),
        'source': 'mw-seasonality'
    }
}

print(json.dumps(result, indent=2))

❌ Fence Execution Error: No input data - mw-seasonality requires 'data' from previous pipeline step Traceback (most recent call last): File "/app/oculus/providers/python_provider.py", line 468, in execute exec(fence_content, exec_globals, exec_locals) File "", line 14, in ValueError: No input data - mw-seasonality requires 'data' from previous pipeline step

Output Contract

The output includes:

• data: Array of records with added fields:- raw_quantity: Original quantity
• seasonality_factor: Factor applied for this week/category
• adjusted_quantity: De-seasonalized quantity (for trend analysis)
• category: Item category used for pattern lookup
• meta: Pipeline metadata for chaining

Usage

# As standalone
seasonality_data = fence('mw-seasonality-fetch')

# Chain to next middleware
trend_input = fence('mw-seasonality-fetch')['data']

South

slots:
- slug: mw-trend
  context:
  - Next step in pipeline - trend calculation
  - 'Pipeline flow: seasonality feeds into trend calculation'

Provenance

Document

• Status: 🔴 Unverified

Changelog

• 2026-01-25 19:08: Node created by mcp - Creating first middleware node for forecast pipeline - seasonality adjuster

East

slots:
- slug: beye-ai-one-pager
  context: []
- slug: wh-demand-history
  context: []
- slug: sprout-garden
  context: []

North

slots:
- slug: wh-demand-history
  context:
  - Linking source data to first middleware step