Analytics & Monitoring

AI Web Feeds provides robust analytics and monitoring capabilities to track feed health, performance, and content trends.

Overview

The analytics system provides:

Summary Metrics - Feed counts, active feed counts, validation success rate, and response times
Trending Topics - Topic activity ranked by feed count and validation frequency
Publication Velocity - Daily publication volume and average articles per day
Snapshots - Daily analytics snapshots stored in the runtime database
CSV Export - Portable analytics extracts for reports and dashboards

Analytics Commands

Summary Metrics

Get a high-level view of all your feeds:

ai-web-feeds analytics summary --date-range 30d

Provides:

Total feeds and active feeds
Validation success rate
Average response time
Feed health distribution

Rank active topics by feed activity:

ai-web-feeds analytics trending --limit 10 --date-range 30d

Shows:

Topic IDs
Feed counts
Validation frequency
Average health score

Publication Velocity

Track publishing volume over a date range:

ai-web-feeds analytics velocity --date-range 30d

Displays:

Total article count
Average articles per day
Daily publication buckets

Quality Scoring

Each feed receives three scores (0-1):

Completeness: How complete is the feed metadata?
Richness: How rich and detailed is the content?
Structure: How well-structured is the feed?

Daily Snapshot

Persist the current analytics snapshot:

ai-web-feeds analytics snapshot

Metrics:

Summary metrics
Trending topic records
Validation aggregates

CSV Export

Export analytics for reports:

ai-web-feeds analytics export --output reports/analytics.csv

Provides:

One row per analytics record
Feed and validation metrics
Spreadsheet-friendly CSV output

Quality Scoring System

Completeness Score (0-1)

Measures how complete the feed metadata is:

✓ Has title ✓ Has description ✓ Has link ✓ Has language ✓ Has timestamps ✓ Has author/publisher ✓ Has canonical topics and raw feed labels ✓ Has image/logo

Richness Score (0-1)

Evaluates content depth and quality:

✓ Items have content ✓ Content coverage percentage ✓ Author attribution ✓ Average content length ✓ Full content availability ✓ Media/images present

Structure Score (0-1)

Assesses feed validity and structure:

✓ No parsing errors ✓ Has items ✓ Items have GUIDs ✓ Has timestamps ✓ Has links

Monitoring Workflows

Daily Health Check

Set up a daily monitoring routine:

#!/bin/bash
# daily-health-check.sh

# Fetch active catalog sources
ai-web-feeds fetch all --limit 100

# Generate health report
ai-web-feeds analytics summary --date-range 30d > daily-summary.txt
ai-web-feeds analytics velocity --date-range 7d > daily-velocity.txt

# Persist a daily snapshot
ai-web-feeds analytics snapshot

Weekly Analytics Review

Generate weekly analytics:

#!/bin/bash
# weekly-analytics.sh

DATE=$(date +%Y-%m-%d)

# Generate CSV report
ai-web-feeds analytics export --output "reports/analytics-${DATE}.csv"

# View trends
ai-web-feeds analytics trending --date-range 7d
ai-web-feeds analytics velocity --date-range 7d

Alert on Feed Failures

Monitor for failing feeds:

#!/bin/bash
# check-failures.sh

# Get summary stats
STATS=$(ai-web-feeds analytics summary --date-range 7d)

# Extract success rate
SUCCESS_RATE=$(echo "$STATS" | grep "Success Rate" | awk '{print $3}' | tr -d '%')

if (( $(echo "$SUCCESS_RATE < 90" | bc -l) )); then
    echo "WARNING: Success rate below 90%: ${SUCCESS_RATE}%"
    # Send alert (email, Slack, etc.)
fi

Advanced Analytics

Custom Python Analysis

Use the Python API for custom analytics:

from ai_web_feeds.analytics import FeedAnalytics
from ai_web_feeds.storage import DatabaseManager

# Initialize
db = DatabaseManager("sqlite:///data/ai-web-feeds.db")
analytics = FeedAnalytics(db.get_session())

# Custom query: Find all feeds with quality < 0.5
feeds = db.get_all_feed_sources()
low_quality = [
    f for f in feeds
    if f.quality_score and f.quality_score < 0.5
]

print(f"Found {len(low_quality)} low quality feeds:")
for feed in low_quality:
    print(f"  - {feed.title}: {feed.quality_score:.2f}")

# Generate custom report
report = analytics.generate_full_report()

# Analyze specific dimension
quality_by_type = {}
for feed in feeds:
    if feed.source_type and feed.quality_score:
        type_name = feed.source_type.value
        if type_name not in quality_by_type:
            quality_by_type[type_name] = []
        quality_by_type[type_name].append(feed.quality_score)

# Calculate averages
for source_type, scores in quality_by_type.items():
    avg = sum(scores) / len(scores)
    print(f"{source_type}: {avg:.3f}")

Database Queries

Direct SQL queries for advanced analysis:

from sqlalchemy import select, func
from ai_web_feeds.models import ArticleEntry, FeedSource

# Get sources with most articles
stmt = (
    select(FeedSource.id, FeedSource.title, func.count(ArticleEntry.id))
    .join(ArticleEntry, ArticleEntry.feed_id == FeedSource.id)
    .group_by(FeedSource.id)
    .order_by(func.count(ArticleEntry.id).desc())
    .limit(10)
)

results = session.exec(stmt).all()
for feed_id, title, count in results:
    print(f"{title}: {count} items")

Export Formats

JSON Reports

Comprehensive analytics in JSON:

{
  "generated_at": "2025-10-15T12:00:00Z",
  "overview": {
    "totals": {
      "feeds": 150,
      "items": 5000,
      "topics": 25
    },
    "feed_status": {
      "verified": 120,
      "active": 100,
      "inactive": 5
    }
  },
  "quality": {
    "average_quality": 0.85,
    "median_quality": 0.87
  }
}

CSV Export (via Python)

import csv
from ai_web_feeds.storage import DatabaseManager

db = DatabaseManager()
feeds = db.get_all_feed_sources()

with open('feeds-export.csv', 'w', newline='') as f:
    writer = csv.writer(f)
    writer.writerow(['ID', 'Title', 'Type', 'Quality', 'Verified'])

    for feed in feeds:
        writer.writerow([
            feed.id,
            feed.title,
            feed.source_type.value if feed.source_type else '',
            feed.quality_score or '',
            feed.verified
        ])

Integration Examples

Grafana Dashboard

Export metrics for Grafana:

import json
from datetime import datetime

def export_metrics():
    analytics = FeedAnalytics(session)
    stats = analytics.get_overview_stats()

    metrics = {
        "timestamp": datetime.utcnow().isoformat(),
        "feeds_total": stats["totals"]["feeds"],
        "feeds_active": stats["feed_status"]["active"],
        "items_24h": stats["recent_activity_24h"]["new_items"]
    }

    with open('/var/lib/grafana/metrics/ai-web-feeds.json', 'w') as f:
        json.dump(metrics, f)

Prometheus Exporter

from prometheus_client import Gauge, generate_latest

feeds_total = Gauge('ai_web_feeds_total', 'Total number of feeds')
feeds_active = Gauge('ai_web_feeds_active', 'Number of active feeds')

def update_metrics():
    stats = analytics.get_overview_stats()
    feeds_total.set(stats["totals"]["feeds"])
    feeds_active.set(stats["feed_status"]["active"])

Best Practices

Regular Monitoring - Run analytics daily to track changes
Health Checks - Monitor feed health scores regularly
Performance Tracking - Watch for degrading fetch success rates
Quality Improvement - Address low-quality feeds
Trend Analysis - Understand publishing patterns
Report Generation - Keep historical analytics for comparison
Alert on Anomalies - Set up alerts for critical issues

CLI Reference - All CLI commands
Python API - Programmatic usage
Database Schema - Data model
Getting Started - Installation guide

Analytics & Monitoring

On this page