Skip to main content

Change Detection Workflow

Learn how to build custom change detection workflows using Meter’s job comparison and history features.

Overview

While Meter’s schedules include automatic change detection, you can build custom workflows for specific needs using job comparison APIs. Use custom change detection when:
  • Need custom change thresholds
  • Comparing non-adjacent jobs
  • Building complex diff logic
  • Generating change reports

Basic comparison

from meter_sdk import MeterClient

client = MeterClient(api_key="sk_live_...")

# Get last two jobs for a strategy
jobs = client.list_jobs(strategy_id="your-strategy-id", limit=2)

if len(jobs) >= 2:
    # Compare latest with previous
    comparison = client.compare_jobs(jobs[0]['id'], jobs[1]['id'])

    print(f"Content hash match: {comparison['content_hash_match']}")
    print(f"Structural match: {comparison['structural_match']}")

    if not comparison['content_hash_match']:
        print("Content has changed!")
        for change in comparison.get('changes', []):
            print(f"  - {change}")

Custom workflows

Track specific field changes

TODO: Implement field-level change detection
def detect_price_changes(strategy_id):
    """Detect price changes specifically"""
    jobs = client.list_jobs(strategy_id=strategy_id, limit=2, status='completed')

    if len(jobs) < 2:
        return []

    old_results = jobs[1]['results']
    new_results = jobs[0]['results']

    changes = []

    # TODO: Implement comparison logic
    # for old, new in zip(old_results, new_results):
    #     if old.get('price') != new.get('price'):
    #         changes.append({
    #             'product': new.get('name'),
    #             'old_price': old.get('price'),
    #             'new_price': new.get('price')
    #         })

    return changes

Generate change reports

TODO: Create change reporting
def generate_change_report(strategy_id, days=7):
    """Generate report of changes over time"""
    history = client.get_strategy_history(strategy_id)

    report = {
        "total_jobs": len(history),
        "jobs_with_changes": sum(1 for j in history if j['has_changes']),
        "change_rate": 0,
        "recent_changes": []
    }

    # TODO: Implement reporting logic
    # Calculate change rate, find patterns, etc.

    return report

Alert on specific changes

TODO: Implement alerting logic
def monitor_with_alerts(strategy_id, alert_threshold=10):
    """Monitor and alert on significant changes"""
    comparison = client.compare_jobs(job_id_1, job_id_2)

    if not comparison['content_hash_match']:
        # TODO: Implement alert logic based on change magnitude
        # if change_magnitude > alert_threshold:
        #     send_alert(...)
        pass

See also

Need help?

Email me at [email protected]