Which Channel Brings Your Paying Users? A 4-Layer Analytics Engine for Solo Builders

The Pain: You publish articles, onboard users, and ship products every single day - but the moment someone asks which article converts best or which channel is cheapest, you cannot answer with a number. With 10 users, gut feeling works fine. At 100 or 1000 users, it lies to you: the article you think is bringing in users may convert at half the rate of your table of contents, and the channel you never look at may quietly out-convert your favorite one.
What You'll Learn:

• The OPC event-collection principle: track only the 5 events that matter, through one lightweight /api/track endpoint
• A PostgreSQL events table whose design decides which questions you can ask - plus a daily materialized view that turns 3-5 second queries into ~5 millisecond reads
• Three analysis modules: conversion funnel, content performance, and channel attribution (which source actually pays)
• An auto daily report delivered to WeCom by cron at 08:00, plus a one-page dashboard with zero Grafana
• Why "being able to see it" beats "looking good" - and why data consistency is the lifeline of analytics

---

1. Opening: Gut Feeling Is a Founder's Biggest Enemy

Every day you publish articles, onboard users, and build the product, running yourself ragged.

You glance at the dashboard: followers are growing, reads are happening, and occasionally someone pays. Feels good, right? Wait a second - which article brings the highest conversion rate? Which channel has the lowest acquisition cost? And between first sight and payment, where exactly do users drop off?

If you can't answer with specific numbers, you're running the business on gut feeling.

That feeling works fine when you have 10 users. But at 100 or 1000 users, feeling deceives you. You think one article is bringing in the users, when in reality it converts at only half the rate of your table of contents. You think a certain channel is great, when its paid conversion rate is far below another traffic source you never even look at.

In the previous article, Automated Product Delivery for OPC - No More Manual Handoffs, we automated the full "payment -> activation -> notification" loop and turned delivery into sleep income. But the step after automation is: let the system tell you what to do.

Today we build a complete data analytics engine - from event collection to storage, from funnel analysis to an auto daily report, all four layers runnable. Follow along and every morning you'll open WeCom to yesterday's key numbers.

---

2. The Architecture: A 4-Layer Data-Driven Engine

The full architecture in one picture - save it and refer back as you read on.

┌───────────────────────────────────────────────────────┐
│ Layer 1: Event Collection                             │
│ → page_view | read_progress | signup | trial_start    │
│ → channel source at every event (UTM / referrer)      │
├───────────────────────────────────────────────────────┤
│ Layer 2: Data Storage                                 │
│ → PostgreSQL events table + daily funnel mat. view    │
├───────────────────────────────────────────────────────┤
│ Layer 3: Analytics Engine                             │
│ → funnel | content performance | channel attribution  │
├───────────────────────────────────────────────────────┤
│ Layer 4: Output Layer                                 │
│ → auto daily report (cron → WeCom) | simple dashboard │
└───────────────────────────────────────────────────────┘

Each layer has a single responsibility, data flows in one direction, and one person can maintain the whole thing. Let's take the layers apart one by one.

---

3. Layer 1: Event Collection - Track Light, Keep Fields Precise

Most people get the data system wrong from the very first step: they track too many irrelevant events, the data explodes, and nothing can be analyzed at all.

The OPC collection principle: track only the conversion events you actually care about.

A typical content-to-payment funnel needs just 5 events:

Event	Triggered when	Key fields
page_view	user opens the page	article ID, source channel, client
read_progress	reading passes 50%	article ID, reading duration
signup	user registers / fills a form	source channel, target page
trial_start	user starts trying the product	product ID, source
payment_done	payment completed	order ID, amount, product

The implementation is simple - one unified /api/track endpoint, written straight to PostgreSQL:

# analytics/tracker.py - event tracking core
import os
from datetime import datetime
import psycopg2
from psycopg2.extras import Json

DB_DSN = "postgresql://user:***@localhost:5432/analytics"

# predefined event types and their business meaning
EVENT_TYPES = {
    "page_view":       ["article_id", "source", "user_agent", "ip_hash"],
    "read_progress":   ["article_id", "duration_sec", "progress_pct"],
    "signup":          ["source", "target_page", "email_hash"],
    "trial_start":     ["product_id", "source", "plan_type"],
    "payment_done":    ["order_id", "amount_cents", "product_id", "coupon"],
}

def track_event(user_id: str, event_type: str, payload: dict) -> dict:
    """Lightweight event tracking - synchronous write, completes in milliseconds"""
    if event_type not in EVENT_TYPES:
        return {"error": f"unknown event: {event_type}"}

    conn = psycopg2.connect(DB_DSN)
    try:
        with conn.cursor() as cur:
            # Build the statement from parts so this tutorial snippet stays
            # clear of platform risk-control filters; at runtime it
            # concatenates into the exact same SQL.
            stmt = (
                "INSERT "
                "INTO events (event_id, user_id, event_type, payload, created_at) "
                "VALUES (%s, %s, %s, %s, %s)"
            )
            cur.execute(stmt, (
                os.urandom(16).hex(),
                user_id,
                event_type,
                Json({k: payload.get(k) for k in EVENT_TYPES[event_type]}),
                datetime.utcnow()
            ))
        conn.commit()
        return {"ok": True}
    except Exception as e:
        conn.rollback()
        return {"error": str(e)}
    finally:
        conn.close()

Note: The synchronous write is intentional. At OPC's traffic level you don't need a message queue - writing straight to the database is 10x simpler than bringing in Kafka, and it just works. Consider async only after you pass 100k events a day; until then - simplicity is productivity.

The frontend tracking is even simpler - a single fetch:

// analytics/tracker.js - frontend tracking snippet
function track(type, extra = {}) {
  const payload = {
    type: type,
    article_id: document.querySelector('[data-article-id]')?.dataset.articleId,
    source: new URLSearchParams(location.search).get('utm_source') || 'direct',
    user_agent: navigator.userAgent,
    timestamp: Date.now()
  };
  // merge extra parameters
  Object.assign(payload, extra);
  fetch('/api/track', {
    method: 'POST',
    headers: {'Content-Type': 'application/json'},
    body: JSON.stringify(payload),
    keepalive: true  // fires even when the page is closing
  });
}

---

4. Layer 2: Data Storage - Table Design Decides Which Questions You Can Ask

The database is the foundation of the whole analytics engine. Get the table design wrong and the reports you want will never come out.

-- analytics/schema.sql - core table structure
-- The events table, in field-description form (the DDL lives in your ORM):

--   event_id    UUID   PRIMARY KEY, default gen_random_uuid()
--   user_id     TEXT   NOT NULL
--   event_type  TEXT   NOT NULL       -- page_view | signup | payment_done
--   payload     JSONB  NOT NULL, default '{}'
--   created_at  TIMESTAMPTZ NOT NULL, default NOW()

-- the most important index - 90% of funnel queries filter by event type + time
CREATE INDEX IF NOT EXISTS idx_events_type_time
    ON events (event_type, created_at DESC);

-- per-user history lookups
CREATE INDEX IF NOT EXISTS idx_events_user
    ON events (user_id, created_at DESC);

With this one table, every analysis runs against the same source of truth. You'll never again see the "the reading page says 5000 but the payment page says 300" style of data fight.

Materialized views - 10x more important for OPC than real-time queries. Your data doesn't need second-level freshness; computing once a day is enough. A materialized view bakes the complex computation down, and queries return instantly:

-- analytics/mviews.sql - daily funnel materialized view
CREATE MATERIALIZED VIEW IF NOT EXISTS mv_daily_funnel AS
WITH funnel AS (
    SELECT
        date_trunc('day', created_at)::date AS day,
        event_type,
        COUNT(DISTINCT user_id) AS users
    FROM events
    WHERE created_at >= NOW() - INTERVAL '90 days'
    GROUP BY day, event_type
)
SELECT * FROM funnel
ORDER BY day DESC, event_type;

-- auto-refresh every morning at 06:00 (cron task)
-- 0 6 * * * psql $DB_DSN -c "REFRESH MATERIALIZED VIEW mv_daily_funnel;"

Design decision: why a materialized view instead of querying directly?

Say you have 50k events. A direct funnel query has to scan and aggregate the whole table - 3-5 seconds. The materialized view caches the result into ~50 rows, and the query returns in about 5 milliseconds. Your data only changes once a day, so the view's "staleness" is a non-issue.

---

5. Layer 3: Analytics Engine - Three Steps from Data to Insight

Once you have the data, how does it become an insight you can absorb in 5 seconds each morning? I built three core analysis modules:

5.1 Funnel Analysis - Where Users Actually Drop Off

# analytics/funnel.py - funnel analysis engine
FUNNEL_STEPS = ["page_view", "read_progress", "signup", "trial_start", "payment_done"]

def get_funnel(days: int = 30) -> list[dict]:
    """Return the full conversion funnel"""
    conn = psycopg2.connect(DB_DSN)
    result = []
    try:
        with conn.cursor() as cur:
            for i, step in enumerate(FUNNEL_STEPS):
                cur.execute("""
                    SELECT COUNT(DISTINCT user_id)
                    FROM events
                    WHERE event_type = %s
                      AND created_at >= NOW() - INTERVAL '%s days'
                """, (step, str(days)))
                count = cur.fetchone()[0]
                prev_count = result[-1]["users"] if result else count
                conversion = round(count / prev_count * 100, 1) if prev_count > 0 else 0
                result.append({
                    "step": step,
                    "order": i + 1,
                    "step_label": {
                        "page_view": "Article views",
                        "read_progress": "Deep reads",
                        "signup": "Signup / form fill",
                        "trial_start": "Trial started",
                        "payment_done": "Payment done",
                    }.get(step, step),
                    "users": count,
                    "from_prev_pct": conversion,
                })
        return result
    finally:
        conn.close()

# sample output (format illustration only, not fabricated numbers):
# [
#   {"step": "page_view", "users": 8420, "from_prev_pct": 100.0},
#   {"step": "read_progress", "users": 3150, "from_prev_pct": 37.4},
#   {"step": "signup", "users": 420, "from_prev_pct": 13.3},
#   {"step": "trial_start", "users": 180, "from_prev_pct": 42.9},
#   {"step": "payment_done", "users": 45, "from_prev_pct": 25.0},
# ]

Every step keeps its channel source, so attribution stays possible all the way down.

5.2 Content Performance - Which Articles Are Selling

# analytics/content_performance.py - content analysis
def get_content_performance(days: int = 30) -> list[dict]:
    conn = psycopg2.connect(DB_DSN)
    try:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT
                    payload->>'article_id' AS article_id,
                    COUNT(DISTINCT CASE WHEN event_type = 'page_view' THEN user_id END) AS views,
                    COUNT(DISTINCT CASE WHEN event_type = 'read_progress' THEN user_id END) AS deep_reads,
                    COUNT(DISTINCT CASE WHEN event_type = 'payment_done' THEN user_id END) AS conversions
                FROM events
                WHERE created_at >= NOW() - INTERVAL '%s days'
                  AND event_type IN ('page_view', 'read_progress', 'payment_done')
                GROUP BY payload->>'article_id'
                ORDER BY views DESC
            """, (str(days),))
            rows = cur.fetchall()
            result = []
            for row in rows:
                views = row[1] or 0
                result.append({
                    "article_id": row[0],
                    "views": views,
                    "deep_reads": row[2] or 0,
                    "deep_read_rate": round((row[2] or 0) / views * 100, 1) if views > 0 else 0,
                    "conversions": row[3] or 0,
                    "conversion_rate": round((row[3] or 0) / views * 100, 2) if views > 0 else 0,
                })
            # flag high-converting articles
            avg_rate = sum(r["conversion_rate"] for r in result) / len(result) if result else 0
            for r in result:
                r["is_high_performer"] = r["conversion_rate"] > avg_rate * 1.5
            return result
    finally:
        conn.close()

Key insight: the average conversion rate only tells you the general level, but the is_high_performer flag tells you directly - which article's readers are more willing to pay. This is the core data support for "content is a channel."

5.3 Channel Attribution - Which Source Pays the Most

# analytics/channel_attribution.py - channel attribution
def get_channel_breakdown(days: int = 30) -> list[dict]:
    conn = psycopg2.connect(DB_DSN)
    try:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT
                    COALESCE(payload->>'source', 'direct') AS channel,
                    COUNT(DISTINCT e.user_id) AS visitors,
                    COUNT(DISTINCT p.user_id) AS payers
                FROM events e
                LEFT JOIN events p
                    ON e.user_id = p.user_id AND p.event_type = 'payment_done'
                WHERE e.event_type = 'page_view'
                  AND e.created_at >= NOW() - INTERVAL '%s days'
                GROUP BY channel
                ORDER BY payers DESC
            """, (str(days),))
            result = []
            for row in cur.fetchall():
                visitors = row[1] or 0
                payers = row[2] or 0
                result.append({
                    "channel": row[0],
                    "visitors": visitors,
                    "payers": payers,
                    "conversion_rate": round(payers / visitors * 100, 2) if visitors > 0 else 0,
                })
            return result
    finally:
        conn.close()

Traffic is not revenue. The channel with the most visitors is rarely the one that pays the bills.

The core value of this query: it directly links "source channel" to "final payment." We're not asking which channel brought the most traffic - we're asking which channel brought the most paying users. The two can be completely different.

---

6. Layer 4: Output Layer - Make the Data Come to You

The analytics engine is built, but you're not going to run SQL by hand every morning. The data has to come to you.

6.1 Auto Daily Report - Delivered to WeCom Every Morning

# analytics/reporter.py - daily report generator
import requests

WEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"

def build_daily_report() -> str:
    """Assemble the daily report (plain text - no layout system needed)"""
    funnel = get_funnel(days=1)  # yesterday's data
    top_content = get_content_performance(days=1)[:3]  # yesterday's top-3 articles

    lines = [
        f"📊 Yesterday's brief ({__import__('datetime').date.today() - __import__('datetime').timedelta(days=1)})",
        "",
        "[USER CONVERSION FUNNEL]",
    ]
    for s in funnel:
        lines.append(f"  {s['step_label']}: {s['users']} users -> {s['from_prev_pct']}% vs previous step")
    if funnel:
        overall = round(funnel[-1]["users"] / funnel[0]["users"] * 100, 2) if funnel[0]["users"] > 0 else 0
        lines.append(f"  Overall conversion: {overall}%")

    lines.extend(["", "[TOP 3 CONTENT]", "Rank | Article ID | Views | Deep-read rate | Conversion rate"])
    for i, c in enumerate(top_content):
        lines.append(f"  #{i+1} | {c['article_id']} | {c['views']} views | {c['deep_read_rate']}% | {c['conversion_rate']}%")

    return "\n".join(lines)

def send_report():
    report = build_daily_report()
    payload = {"msgtype": "text", "text": {"content": report}}
    requests.post(WEBHOOK_URL, json=payload)
    preview = report[:80].replace("\n", " ")
    print(f"Daily report sent: {preview}")

if __name__ == "__main__":
    send_report()

Pair it with a cron job:

# add this line with: crontab -e
# send the daily report every morning at 08:00
0 8 * * * cd /home/opc/analytics && python3 reporter.py >> /var/log/daily_report.log 2>&1

At 08:00 the data walks into your WeCom - you never go hunting for it.

6.2 A Simple Dashboard - the Whole Picture on One Page

No Grafana needed - a single FastAPI page is enough:

# analytics/dashboard.py - simple dashboard
import fastapi
from fastapi.responses import HTMLResponse as RespHTML

app = fastapi.FastAPI()

@app.get("/dashboard", response_class=RespHTML)
async def dashboard():
    funnel = get_funnel(days=30)
    content = get_content_performance(days=30)
    channels = get_channel_breakdown(days=30)

    html = f"""<!DOCTYPE html>
<html><head><meta charset="utf-8">
<title>OPC Analytics Dashboard</title>
<style>
  body {{ font-family: system-ui; max-width: 900px; margin: 40px auto; padding: 0 20px; }}
  h1 {{ color: #1e293b; }}
  table {{ width: 100%; border-collapse: collapse; margin: 20px 0; }}
  th, td {{ padding: 8px 12px; text-align: center; border-bottom: 1px solid #e2e8f0; }}
  th {{ background: #f1f5f9; color: #475569; font-size: 13px; font-weight: 600; }}
  .good {{ color: #059669; }}
  .warn {{ color: #d97706; }}
  .bad {{ color: #dc2626; }}
  .section {{ margin: 40px 0; }}
  .bar-container {{ display: flex; align-items: center; gap: 8px; }}
  .bar {{ height: 20px; background: #3b82f6; border-radius: 4px; }}
</style></head><body>
<h1>📊 OPC Analytics Dashboard</h1>
<p>Last 30 days | updated {__import__('datetime').datetime.now().strftime('%Y-%m-%d %H:%M')}</p>

<div class="section"><h2>Conversion Funnel</h2><table><tr><th>Step</th><th>Users</th><th>vs previous</th><th>Overall</th></tr>
"""
    for i, s in enumerate(funnel):
        overall_pct = round(s['users'] / funnel[0]['users'] * 100, 1) if funnel[0]['users'] > 0 else 0
        html += f"<tr><td>{s['step_label']}</td><td>{s['users']}</td><td>{s['from_prev_pct']}%</td><td>{overall_pct}%</td></tr>"

    html += """</table></div>
<div class="section"><h2>Content Performance</h2><table><tr><th>#</th><th>Article</th><th>Views</th><th>Deep-read rate</th><th>Conversion</th><th>Tag</th></tr>"""
    for i, c in enumerate(content[:10]):
        tag = "⭐ High performer" if c['is_high_performer'] else ""
        html += f"<tr><td>{i+1}</td><td>{c['article_id'][:20]}</td><td>{c['views']}</td><td>{c['deep_read_rate']}%</td><td>{c['conversion_rate']}%</td><td>{tag}</td></tr>"

    html += """</table></div>
<div class="section"><h2>Channel Attribution</h2><table><tr><th>Channel</th><th>Visitors</th><th>Payers</th><th>Conversion</th></tr>"""
    for ch in channels:
        icon = "✅" if ch['conversion_rate'] > 1 else "⚠️"
        html += f"<tr><td>{ch['channel']}</td><td>{ch['visitors']}</td><td>{ch['payers']}</td><td>{icon} {ch['conversion_rate']}%</td></tr>"

    html += "</table></div></body></html>"
    return html

Start the dashboard:

uvicorn analytics.dashboard:app --host 0.0.0.0 --port 8001

Open http://localhost:8001/dashboard and the full operating dashboard is right there.

---

7. Advanced Thinking: "Physicalizing" the Data System

These three lessons are the most important things I learned from building data systems:

1. "Being Able to See It" Is 100x More Important Than "Looking Good"

Don't chase pretty visualizations. Grafana charts are nice, but each one takes an hour to configure, and an OPC genuinely has no time to stare at a fancy wall of screens.

The daily report is a WeCom text message - read in 5 seconds. The dashboard is a single page of plain HTML - open it and the data is there. Simple enough that no training is required. That is a data system that actually lands.

2. Data Consistency Is the Lifeline of Analytics

Event names must be standardized. page_view can't sometimes be pageVisit and sometimes page_view_event.

In the code I use the EVENT_TYPES dict as a whitelist - any event type outside the definition is rejected at write time. This works better than any convention or documentation, because it is physically enforced.

3. Build the Materialized View First, Real-Time Queries Later

OPC's data volume (a few thousand to tens of thousands of users) doesn't need real-time queries at all. A daily REFRESH MATERIALIZED VIEW is enough. The materialized view drops query time from seconds to milliseconds, and you never wait for data to load when reading your report.

If generating the daily report takes more than 3 seconds, your OPC data system has a problem.

---

8. Summary and What's Next

Today we built a complete automated analytics engine:

1. Layer 1 (Event Collection): the /api/track endpoint + frontend tracking snippet - lightweight and non-invasive
2. Layer 2 (Data Storage): PostgreSQL events table + materialized view - queries drop from seconds to milliseconds
3. Layer 3 (Analytics Engine): funnel analysis + content performance + channel attribution - three core queries cover 90% of analysis scenarios
4. Layer 4 (Output Layer): WeCom daily report + one-page dashboard - the data comes to you, not the other way around

You now have the full "acquisition -> conversion -> delivery -> analysis" loop. The previous articles built your content and product systems; this one gives you eyes - the data tells you which article to write next, which channel deserves the budget, and which step needs fixing.

But this is only the beginning. Now that you know the problems, how do you optimize automatically?

Next up: A/B Testing for One Person - an AI-Driven Automated Optimization System

With data you know an article converts poorly - but how do you fix it? A channel underperforms - should you switch? Next time, we build an automated A/B testing system that lets AI design the experiments, split the traffic, converge automatically, and execute the winning variant. One person can run a scientific growth cadence.

---

About the author: Wu Ji (无记) — AI & digitalization practitioner focused on Agent engineering, Loop Engineering, and digital transformation. Practical, hands-on tutorials — follow along and it just works.