Data Freshness Patterns

The Danish Parliament API provides exceptional data freshness with real-time parliamentary activity reflected within hours. This document analyzes update patterns, monitoring strategies, and practical guidance for leveraging the API's real-time capabilities.

Real-Time Update Characteristics

Update Frequency Analysis

Based on comprehensive Phase 29 investigation findings:

UPDATE_FREQUENCY_PATTERNS = {
    'current_data': {
        'latest_case_update': '2025-09-09T17:49:11.87',
        'latest_actor_update': '2025-09-09T17:29:09.407', 
        'latest_voting_update': '2025-09-09T12:30:12.467',
        'investigation_time': '2025-09-09T21:15:00',
        'freshness_assessment': 'EXTREMELY_FRESH (3-8 hours)'
    },

    'daily_volume': {
        'average_daily_updates': 55,
        'todays_updates': 59,
        'yesterdays_updates': 46,
        'year_to_date_total': 3782,
        'assessment': 'ACTIVE_DAILY_PROCESSING'
    },

    'business_hours_patterns': {
        'peak_update_window': '16:00-17:30 Danish time',
        'voting_session_updates': '12:30 (midday sessions)',
        'batch_processing_time': '17:29 (end-of-day)',
        'low_activity_periods': '07:00-09:00, 19:00-23:00'
    }
}

Entity-Specific Freshness Profiles

class EntityFreshnessAnalyzer:
    def __init__(self):
        # Based on real API testing and observation
        self.freshness_profiles = {
            'Sag': {
                'typical_update_lag': '2-6 hours',
                'max_observed_lag': '24 hours',
                'update_triggers': [
                    'Status changes (new ’ review ’ voting ’ approved)',
                    'Document attachment completion',
                    'Committee assignment changes',
                    'Voting session completion',
                    'Administrative updates'
                ],
                'batch_indicators': 'Multiple related cases updated simultaneously',
                'monitoring_field': 'opdateringsdato',
                'peak_activity': 'Parliamentary session days'
            },

            'Aktør': {
                'typical_update_lag': '4-8 hours',
                'max_observed_lag': '48 hours',
                'update_triggers': [
                    'Role changes (minister appointments, resignations)',
                    'Committee membership changes',
                    'Contact information updates',
                    'Biographical information additions',
                    'Administrative corrections'
                ],
                'batch_indicators': '15 actors updated at 17:29:09.407',
                'monitoring_field': 'opdateringsdato',
                'peak_activity': 'Government formation periods, election aftermath'
            },

            'Afstemning': {
                'typical_update_lag': '1-3 hours',
                'max_observed_lag': '12 hours',
                'update_triggers': [
                    'Voting session completion',
                    'Vote result certification',
                    'Vote count finalization',
                    'Quorum validation'
                ],
                'batch_indicators': 'Voting sessions processed as complete units',
                'monitoring_field': 'opdateringsdato',
                'peak_activity': 'Parliamentary voting days'
            },

            'Stemme': {
                'typical_update_lag': '1-4 hours',
                'max_observed_lag': '12 hours',
                'update_triggers': [
                    'Individual vote recording',
                    'Absentee vote processing',
                    'Vote correction submissions'
                ],
                'batch_indicators': 'All votes from session updated together',
                'monitoring_field': 'opdateringsdato', 
                'peak_activity': 'Immediately following voting sessions'
            },

            'Dokument': {
                'typical_update_lag': '6-12 hours',
                'max_observed_lag': '72 hours',
                'update_triggers': [
                    'Document upload completion',
                    'PDF generation completion',
                    'Metadata assignment',
                    'Document categorization',
                    'Access permission finalization'
                ],
                'batch_indicators': 'Related documents processed together',
                'monitoring_field': 'opdateringsdato',
                'peak_activity': 'Following committee meetings, document deadlines'
            },

            'Møde': {
                'typical_update_lag': '12-24 hours',
                'max_observed_lag': '72 hours', 
                'update_triggers': [
                    'Meeting completion',
                    'Agenda finalization',
                    'Participant list confirmation',
                    'Document attachment',
                    'Minutes completion'
                ],
                'batch_indicators': 'Meeting-related records updated as set',
                'monitoring_field': 'opdateringsdato',
                'peak_activity': '24-48 hours post-meeting'
            }
        }

    def get_freshness_expectations(self, entity_type, parliamentary_context):
        """Get freshness expectations for specific context"""

        profile = self.freshness_profiles.get(entity_type, {})

        # Adjust expectations based on parliamentary context
        context_adjustments = {
            'active_session': {
                'multiplier': 0.5,  # Faster updates during active sessions
                'note': 'Real-time processing during active parliamentary periods'
            },
            'recess': {
                'multiplier': 2.0,  # Slower updates during recess
                'note': 'Reduced processing frequency during parliamentary recess'
            },
            'weekend': {
                'multiplier': 3.0,  # Much slower weekend processing
                'note': 'Minimal automated processing only'
            },
            'holiday': {
                'multiplier': 5.0,  # Very slow during holidays
                'note': 'Emergency processing only'
            }
        }

        adjustment = context_adjustments.get(parliamentary_context, {'multiplier': 1.0})

        return {
            'entity_type': entity_type,
            'base_profile': profile,
            'context': parliamentary_context,
            'adjusted_expectations': {
                'typical_lag_hours': profile.get('typical_update_lag', 'Unknown'),
                'context_multiplier': adjustment['multiplier'],
                'context_note': adjustment.get('note')
            }
        }

Update Pattern Detection

Batch Processing Analysis

class BatchProcessingDetector:
    def __init__(self):
        # Patterns observed during investigation
        self.known_batch_patterns = {
            'actor_batch_2025_09_09': {
                'timestamp': '2025-09-09T17:29:09.407',
                'entities_affected': 15,
                'entity_type': 'Aktør',
                'pattern_type': 'End-of-day administrative batch',
                'indicators': 'Identical timestamps to millisecond precision'
            },

            'voting_session_batches': {
                'pattern': 'All votes from session updated simultaneously',
                'entity_types': ['Afstemning', 'Stemme'],
                'frequency': 'Per voting session',
                'detection_method': 'Group by afstemningid + timestamp'
            },

            'document_processing_batches': {
                'pattern': 'Related documents processed together',
                'entity_types': ['Dokument', 'Fil'],
                'frequency': 'Following committee meetings',
                'detection_method': 'Group by sagid + timestamp'
            }
        }

    def detect_batch_processing(self, entity_type, time_window_hours=4):
        """Detect batch processing patterns in recent updates"""

        # Query for recent updates
        from datetime import datetime, timedelta
        cutoff_time = datetime.now() - timedelta(hours=time_window_hours)

        params = {
            "$filter": f"opdateringsdato gt datetime'{cutoff_time.isoformat()}'",
            "$orderby": "opdateringsdato desc",
            "$top": 100
        }

        response = requests.get(f"https://oda.ft.dk/api/{entity_type}", params=params)

        if response.status_code == 200:
            records = response.json().get('value', [])

            # Group by timestamp
            timestamp_groups = {}
            for record in records:
                timestamp = record.get('opdateringsdato')
                if timestamp:
                    if timestamp not in timestamp_groups:
                        timestamp_groups[timestamp] = []
                    timestamp_groups[timestamp].append(record)

            # Identify batches (multiple records with same timestamp)
            batches = {
                ts: records for ts, records in timestamp_groups.items() 
                if len(records) > 1
            }

            return {
                'detection_window': f"{time_window_hours} hours",
                'total_recent_updates': len(records),
                'batch_timestamps': len(batches),
                'largest_batch_size': max(len(r) for r in batches.values()) if batches else 0,
                'batch_details': [
                    {
                        'timestamp': ts,
                        'record_count': len(records),
                        'record_ids': [r.get('id') for r in records[:5]]  # First 5 IDs
                    }
                    for ts, records in list(batches.items())[:3]  # Top 3 batches
                ]
            }

        return {'error': 'Unable to fetch recent updates for batch analysis'}

Real-Time Monitoring Implementation

class RealTimeMonitor:
    def __init__(self, entities_to_monitor=None):
        self.entities = entities_to_monitor or ['Sag', 'Aktør', 'Afstemning', 'Møde']
        self.last_check_timestamps = {}

    def initialize_monitoring(self):
        """Initialize monitoring by getting current timestamps"""
        for entity in self.entities:
            latest_update = self._get_latest_update(entity)
            if latest_update:
                self.last_check_timestamps[entity] = latest_update

        return {
            'monitoring_initialized': True,
            'entities_monitored': len(self.entities),
            'baseline_timestamps': self.last_check_timestamps
        }

    def check_for_updates(self, entity_type=None):
        """Check for updates since last monitoring run"""

        entities_to_check = [entity_type] if entity_type else self.entities
        update_summary = {}

        for entity in entities_to_check:
            last_known_update = self.last_check_timestamps.get(entity)

            if last_known_update:
                # Query for records updated since last check
                params = {
                    "$filter": f"opdateringsdato gt datetime'{last_known_update}'",
                    "$orderby": "opdateringsdato desc",
                    "$select": "id,opdateringsdato",
                    "$top": 50
                }

                response = requests.get(f"https://oda.ft.dk/api/{entity}", params=params)

                if response.status_code == 200:
                    new_updates = response.json().get('value', [])

                    if new_updates:
                        # Update our timestamp
                        self.last_check_timestamps[entity] = new_updates[0]['opdateringsdato']

                        update_summary[entity] = {
                            'new_updates_found': len(new_updates),
                            'latest_update': new_updates[0]['opdateringsdato'],
                            'updated_record_ids': [r['id'] for r in new_updates[:10]]
                        }
                    else:
                        update_summary[entity] = {
                            'new_updates_found': 0,
                            'latest_update': last_known_update
                        }

        return {
            'check_timestamp': datetime.now().isoformat(),
            'entities_checked': len(entities_to_check),
            'updates_found': sum(
                s.get('new_updates_found', 0) for s in update_summary.values()
            ),
            'entity_updates': update_summary
        }

    def _get_latest_update(self, entity_type):
        """Get the most recent update timestamp for entity"""

        response = requests.get(
            f"https://oda.ft.dk/api/{entity_type}",
            params={
                "$orderby": "opdateringsdato desc",
                "$select": "opdateringsdato",
                "$top": 1
            }
        )

        if response.status_code == 200:
            data = response.json().get('value', [])
            if data:
                return data[0].get('opdateringsdato')

        return None

    def get_update_frequency_stats(self, entity_type, days_back=7):
        """Analyze update frequency over recent period"""

        from datetime import datetime, timedelta
        cutoff_date = datetime.now() - timedelta(days=days_back)

        params = {
            "$filter": f"opdateringsdato gt datetime'{cutoff_date.isoformat()}'",
            "$select": "opdateringsdato",
            "$orderby": "opdateringsdato desc",
            "$top": 1000
        }

        response = requests.get(f"https://oda.ft.dk/api/{entity_type}", params=params)

        if response.status_code == 200:
            updates = response.json().get('value', [])

            # Group by day
            daily_counts = {}
            for update in updates:
                update_date = update['opdateringsdato'][:10]  # YYYY-MM-DD
                daily_counts[update_date] = daily_counts.get(update_date, 0) + 1

            # Calculate statistics
            daily_values = list(daily_counts.values())
            avg_daily = sum(daily_values) / len(daily_values) if daily_values else 0

            return {
                'analysis_period_days': days_back,
                'total_updates': len(updates),
                'days_with_activity': len(daily_counts),
                'average_daily_updates': round(avg_daily, 1),
                'max_daily_updates': max(daily_values) if daily_values else 0,
                'min_daily_updates': min(daily_values) if daily_values else 0,
                'daily_breakdown': dict(sorted(daily_counts.items(), reverse=True)[:7])
            }

        return {'error': 'Unable to analyze update frequency'}

Practical Monitoring Strategies

Change Detection Implementation

class ChangeDetectionSystem:
    def __init__(self):
        self.polling_strategies = {
            'real_time': {
                'interval_seconds': 300,  # 5 minutes
                'use_case': 'Live dashboards, breaking news',
                'cost': 'HIGH - Many API calls',
                'accuracy': 'MAXIMUM'
            },
            'near_real_time': {
                'interval_seconds': 900,  # 15 minutes  
                'use_case': 'News monitoring, alerts',
                'cost': 'MEDIUM',
                'accuracy': 'HIGH'
            },
            'regular': {
                'interval_seconds': 3600,  # 1 hour
                'use_case': 'Research updates, daily summaries',
                'cost': 'LOW',
                'accuracy': 'GOOD'
            },
            'batch': {
                'interval_seconds': 86400,  # 24 hours
                'use_case': 'Historical analysis, archives',
                'cost': 'VERY_LOW',
                'accuracy': 'SUFFICIENT'
            }
        }

    def implement_change_detection(self, strategy🔧near_real_time', entities=None):
        """Implement change detection with specified strategy"""

        strategy_config = self.polling_strategies.get(strategy)
        if not strategy_config:
            raise ValueError(f"Unknown strategy: {strategy}")

        entities = entities or ['Sag', 'Afstemning', 'Aktør']

        implementation_plan = {
            'strategy': strategy,
            'configuration': strategy_config,
            'entities_monitored': entities,
            'estimated_api_calls_per_day': len(entities) * (86400 / strategy_config['interval_seconds']),
            'implementation': self._generate_implementation_code(strategy, entities)
        }

        return implementation_plan

    def _generate_implementation_code(self, strategy, entities):
        """Generate implementation code for change detection"""

        code_template = f'''
import time
import requests
from datetime import datetime

class DanishParliamentChangeDetector:
    def __init__(self):
        self.entities = {entities}
        self.last_timestamps = {{}}
        self.polling_interval = {self.polling_strategies[strategy]["interval_seconds"]}

    def start_monitoring(self):
        """Start continuous monitoring"""
        print(f"Starting {{self.__class__.__name__}} with {strategy} strategy")

        while True:
            changes = self.check_for_changes()
            if changes['total_changes'] > 0:
                self.handle_changes(changes)

            time.sleep(self.polling_interval)

    def check_for_changes(self):
        """Check all entities for changes"""
        all_changes = {{}}
        total_changes = 0

        for entity in self.entities:
            entity_changes = self._check_entity_changes(entity)
            all_changes[entity] = entity_changes
            total_changes += len(entity_changes.get('new_records', []))

        return {{
            'check_time': datetime.now().isoformat(),
            'total_changes': total_changes,
            'entity_changes': all_changes
        }}

    def _check_entity_changes(self, entity_type):
        """Check specific entity for changes"""
        last_timestamp = self.last_timestamps.get(entity_type)

        params = {{"$orderby": "opdateringsdato desc", "$top": 50}}

        if last_timestamp:
            params["$filter"] = f"opdateringsdato gt datetime'{{last_timestamp}}'"

        response = requests.get(f"https://oda.ft.dk/api/{{entity_type}}", params=params)

        if response.status_code == 200:
            new_records = response.json().get('value', [])

            if new_records:
                # Update timestamp for next check
                self.last_timestamps[entity_type] = new_records[0]['opdateringsdato']

            return {{
                'entity': entity_type,
                'new_records': new_records,
                'count': len(new_records)
            }}

        return {{'entity': entity_type, 'new_records': [], 'count': 0, 'error': True}}

    def handle_changes(self, changes):
        """Handle detected changes"""
        print(f"Changes detected: {{changes['total_changes']}} total")

        for entity, entity_changes in changes['entity_changes'].items():
            if entity_changes['count'] > 0:
                print(f"  {{entity}}: {{entity_changes['count']}} updates")

                # Add your change handling logic here:
                # - Send notifications
                # - Update databases  
                # - Trigger downstream processing
                # - Generate alerts
'''

        return code_template

Performance Optimization for Monitoring

class OptimizedFreshnessChecker:
    def __init__(self):
        self.efficient_queries = {
            'timestamp_only_check': {
                'params': {'$select': 'opdateringsdato', '$top': 1, '$orderby': 'opdateringsdato desc'},
                'purpose': 'Quick freshness check without data download',
                'bandwidth_usage': 'MINIMAL'
            },

            'summary_check': {
                'params': {'$select': 'id,opdateringsdato', '$top': 10, '$orderby': 'opdateringsdato desc'},
                'purpose': 'Identify recently changed records',
                'bandwidth_usage': 'LOW'
            },

            'delta_sync': {
                'params': {'$filter': 'opdateringsdato gt datetime\'[TIMESTAMP]\'', '$top': 100},
                'purpose': 'Incremental sync of changes',
                'bandwidth_usage': 'VARIABLE'
            }
        }

    def check_freshness_efficiently(self, entity_type, check_type🔧timestamp_only_check'):
        """Perform efficient freshness check"""

        query_config = self.efficient_queries.get(check_type)
        if not query_config:
            raise ValueError(f"Unknown check type: {check_type}")

        start_time = time.time()

        response = requests.get(
            f"https://oda.ft.dk/api/{entity_type}",
            params=query_config['params']
        )

        request_time = time.time() - start_time

        if response.status_code == 200:
            data = response.json()
            records = data.get('value', [])

            return {
                'entity_type': entity_type,
                'check_type': check_type,
                'records_returned': len(records),
                'response_time_seconds': round(request_time, 3),
                'response_size_bytes': len(response.content),
                'latest_update': records[0].get('opdateringsdato') if records else None,
                'efficiency_rating': self._rate_efficiency(request_time, len(response.content))
            }

        return {
            'entity_type': entity_type,
            'check_type': check_type,
            'error': f"HTTP {response.status_code}",
            'response_time_seconds': round(request_time, 3)
        }

    def _rate_efficiency(self, response_time, response_size):
        """Rate the efficiency of the query"""

        if response_time < 0.2 and response_size < 5000:
            return 'EXCELLENT'
        elif response_time < 0.5 and response_size < 20000:
            return 'GOOD'
        elif response_time < 1.0:
            return 'ACCEPTABLE'
        else:
            return 'INEFFICIENT'

Best Practices for Real-Time Applications

Polling Strategy Selection

def select_polling_strategy(use_case, update_criticality, resource_budget):
    """Select appropriate polling strategy based on requirements"""

    strategy_matrix = {
        ('breaking_news', 'CRITICAL', 'HIGH'): {
            'strategy': 'real_time',
            'interval': 300,  # 5 minutes
            'justification': 'Maximum freshness for critical news'
        },
        ('research_monitoring', 'MEDIUM', 'MEDIUM'): {
            'strategy': 'near_real_time', 
            'interval': 900,  # 15 minutes
            'justification': 'Good balance of freshness and efficiency'
        },
        ('historical_analysis', 'LOW', 'LOW'): {
            'strategy': 'batch',
            'interval': 86400,  # Daily
            'justification': 'Sufficient for non-time-sensitive analysis'
        },
        ('dashboard_display', 'MEDIUM', 'MEDIUM'): {
            'strategy': 'regular',
            'interval': 3600,  # Hourly
            'justification': 'Reasonable freshness for public displays'
        }
    }

    key = (use_case, update_criticality, resource_budget)
    recommendation = strategy_matrix.get(key)

    if not recommendation:
        # Fallback logic
        if update_criticality == 'CRITICAL':
            recommendation = {'strategy': 'near_real_time', 'interval': 900}
        elif resource_budget == 'LOW':
            recommendation = {'strategy': 'batch', 'interval': 86400}
        else:
            recommendation = {'strategy': 'regular', 'interval': 3600}

    return recommendation

Error Handling for Real-Time Monitoring

class RobustMonitoring:
    def __init__(self):
        self.error_handling_strategies = {
            'network_timeout': 'Exponential backoff with max 5 minute delay',
            'rate_limiting': 'Implement client-side throttling',  
            'api_downtime': 'Fallback to cached data with staleness warnings',
            'invalid_response': 'Log error and skip polling cycle',
            'authentication_error': 'Not applicable - no auth required'
        }

    def monitor_with_resilience(self, entity_type, max_retries=3):
        """Monitor with robust error handling"""

        for attempt in range(max_retries):
            try:
                result = self._attempt_monitoring(entity_type)

                if result.get('success'):
                    return result
                else:
                    if attempt < max_retries - 1:
                        # Exponential backoff
                        delay = 2 ** attempt
                        time.sleep(delay)
                        continue

            except requests.exceptions.Timeout:
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)
                    continue
                else:
                    return {'success': False, 'error': 'Timeout after retries'}

            except requests.exceptions.RequestException as e:
                return {'success': False, 'error': f'Network error: {e}'}

        return {'success': False, 'error': 'Max retries exceeded'}

Freshness Monitoring Dashboard

Key Metrics to Track

FRESHNESS_DASHBOARD_METRICS = {
    'primary_indicators': [
        'Time since last update (per entity)',
        'Average update lag (24-hour rolling)',
        'Update frequency (updates per hour)',
        'Batch processing detection'
    ],

    'secondary_indicators': [
        'Parliamentary activity correlation',
        'Weekend/holiday update patterns', 
        'Peak activity periods identification',
        'Update prediction modeling'
    ],

    'alert_thresholds': {
        'stale_data_warning': '12 hours without updates',
        'stale_data_critical': '48 hours without updates',
        'unusual_batch_size': '>50 records same timestamp',
        'update_frequency_drop': '<10 updates in 24 hours'
    }
}

Integration with Parliamentary Schedule

Understanding Update Context

def correlate_updates_with_parliamentary_activity():
    """Correlate API updates with known parliamentary schedule"""

    # This would integrate with parliamentary calendar
    correlation_patterns = {
        'voting_days': {
            'expected_entities': ['Afstemning', 'Stemme', 'Sag'],
            'expected_lag': '1-3 hours post-voting',
            'volume_multiplier': 3.0
        },

        'committee_days': {
            'expected_entities': ['Møde', 'Dokument', 'SagAktør'],
            'expected_lag': '12-24 hours post-meeting',
            'volume_multiplier': 1.5
        },

        'recess_periods': {
            'expected_entities': ['Aktør', 'Dokument'],
            'expected_lag': '24-72 hours',
            'volume_multiplier': 0.3
        }
    }

    return correlation_patterns

The Danish Parliament API's exceptional freshness characteristics make it suitable for real-time applications, news monitoring, and live political analysis. Understanding update patterns enables efficient monitoring strategies while respecting the API infrastructure and maintaining reliable application performance.