metricsfirst 0.3.0

MetricsFirst SDK for Python - Analytics for Telegram bots and web funnels

MetricsFirst Python SDK

Official Python SDK for MetricsFirst - Analytics for Telegram bots and web funnels.

Note: Commands and interactions are tracked automatically when you add your bot to MetricsFirst. This SDK is for tracking custom events like services, purchases, errors, and funnel events.

Installation

# Basic installation (sync only)
pip install metricsfirst

# With async support
pip install metricsfirst[async]

Features

• Fire-and-forget: All tracking calls are non-blocking and don't add latency
• Background sending: Events are sent in a separate thread (sync) or task (async)
• Auto cleanup: No need to call shutdown() - events flush automatically on exit
• Error resilience: Errors are logged, never thrown to your code
• Memory safe: Queue is limited to 1000 events to prevent memory issues
• Custom Events: Track any event with dynamic properties (Mixpanel-style)
• Funnel Tracking: Track anonymous visitors before registration with ad_id

Quick Start

Custom Events (Mixpanel-style)

Track any event with dynamic properties:

from metricsfirst import MetricsFirst

# Initialize once (globally)
mf = MetricsFirst(
    bot_id="your_bot_id",
    api_key="your_api_key",
)

# Track custom events with any properties
mf.track(123456789, 'STORY_RESPONSE', {
    'target': 'username123',
    'url': 'https://example.com/story',
    'response_time_ms': 150,
    'success': True,
})

mf.track(123456789, 'BUTTON_CLICK', {
    'button_name': 'premium_upgrade',
    'screen': 'main_menu',
})

mf.track(123456789, 'VIDEO_DOWNLOADED', {
    'duration_seconds': 45,
    'quality': '1080p',
    'source': 'instagram',
})

# Events are automatically flushed on exit - no shutdown() needed!

Synchronous Client

from metricsfirst import MetricsFirst, ServiceEventData

# Initialize once (globally)
mf = MetricsFirst(
    bot_id="your_bot_id",
    api_key="your_api_key",
)

# Track a service (fire-and-forget, non-blocking)
mf.track_service(ServiceEventData(
    user_id=123456789,
    service_name="image_generation",
    is_free=False,
    price=10,
    currency="USD",
))

# Events are automatically flushed on exit

Asynchronous Client

import asyncio
from metricsfirst import AsyncMetricsFirst, ServiceEventData

# Initialize once (globally)
mf = AsyncMetricsFirst(
    bot_id="your_bot_id",
    api_key="your_api_key",
)

async def main():
    # Just use it - auto-starts on first call
    await mf.track_service(ServiceEventData(
        user_id=123456789,
        service_name="image_generation",
    ))

# Events are automatically flushed on exit
asyncio.run(main())

Context Manager

# Sync
with MetricsFirst(bot_id="...", api_key="...") as mf:
    mf.track_service(...)

# Async
async with AsyncMetricsFirst(bot_id="...", api_key="...") as mf:
    await mf.track_service(...)

Funnel Tracking

Track anonymous visitors before they register. Perfect for landing pages and ad attribution.

The Flow

1. Landing page: User visits with UTM params → track with ad_id and funnel_id
2. CTA click: User clicks button → track with same ad_id
3. Bot start: User registers → link ad_id to user_id
4. Onwards: Track with user_id

Example (Recommended: Funnel ID Mode)

Using track_funnel_step with a funnel_id enables automatic step discovery in the dashboard:

from metricsfirst import MetricsFirst, UtmParams

mf = MetricsFirst(api_key='your_api_key')
FUNNEL_ID = 'my_landing_funnel'

# 1. In your web backend: track anonymous visitor with funnel_id
mf.track_funnel_step(
    funnel_id=FUNNEL_ID,
    step='page_view',
    ad_id='visitor_abc123',
    properties={'page': '/landing', 'referrer': request.headers.get('Referer')},
    utm=UtmParams(
        utm_source='google',
        utm_medium='cpc',
        utm_campaign='spring_sale',
    ),
)

# 2. CTA click (from landing page)
mf.track_funnel_step(
    funnel_id=FUNNEL_ID,
    step='cta_click',
    ad_id='visitor_abc123',
    properties={'button': 'start_bot', 'position': 'hero'},
)

# 3. In your bot: link ad_id to user_id when user starts
# The ad_id is passed via start parameter: t.me/bot?start=visitor_abc123
ad_id = extract_from_start_param(update.message.text)
if ad_id:
    mf.link_user_identity(
        ad_id=ad_id,
        user_id=update.effective_user.id,
        properties={'registration_source': 'landing_cta'},
    )

# 4. Now track with user_id as usual
mf.track(update.effective_user.id, 'subscription_started', {'plan': 'pro'})

Async Example

from metricsfirst import AsyncMetricsFirst, UtmParams

mf = AsyncMetricsFirst(api_key='your_api_key')
FUNNEL_ID = 'my_landing_funnel'

# Track anonymous visitor with funnel_id
await mf.track_funnel_step(
    funnel_id=FUNNEL_ID,
    step='page_view',
    ad_id='visitor_abc123',
    utm=UtmParams(utm_source='google', utm_campaign='spring'),
)

# Link to user when they register
await mf.link_user_identity('visitor_abc123', user_id=123456789)

Legacy Example

# Track without funnel_id (requires manual step configuration in dashboard)
mf.track_funnel(
    ad_id='visitor_abc123',
    event_name='page_view',
    properties={'page': '/landing'},
    utm=UtmParams(utm_source='google', utm_campaign='spring'),
)

# Link identity (legacy method)
mf.link_identity(ad_id='visitor_abc123', user_id=123456789)

Methods

# RECOMMENDED: Track funnel step with funnel_id (auto-discovery)
mf.track_funnel_step(
    funnel_id: str,              # Funnel identifier (e.g., 'landing_funnel')
    step: str,                   # Step name (e.g., 'page_view', 'cta_click')
    ad_id: str,                  # Anonymous visitor ID
    properties: dict = None,     # Event properties
    utm: UtmParams = None,       # UTM parameters
)

# RECOMMENDED: Link ad_id to user_id with explicit event
mf.link_user_identity(
    ad_id: str,                  # Anonymous visitor ID
    user_id: int,                # Telegram user ID
    properties: dict = None,     # Additional context
)

# Legacy: Track funnel event (requires manual dashboard configuration)
mf.track_funnel(
    ad_id: str,                  # Anonymous visitor ID
    event_name: str,             # Event name
    properties: dict = None,     # Event properties
    utm: UtmParams = None,       # UTM parameters
)

# Legacy: Link anonymous ad_id to registered user_id
mf.link_identity(
    ad_id: str,                  # Anonymous visitor ID
    user_id: int,                # Telegram user ID
    properties: dict = None,     # Additional context
)

Available Methods

Method	Description
track()	Track custom events with any properties
track_funnel_step()	Track funnel step with funnel_id (recommended)
link_user_identity()	Link ad_id to user_id (recommended)
track_funnel()	Track anonymous funnel events (legacy)
link_identity()	Link ad_id to user_id (legacy)
track_service()	Track services provided
track_error()	Track errors
track_error_from_exception()	Track error from exception
track_purchase_initiated()	Track purchase start
track_purchase_completed()	Track successful purchase
track_purchase_error()	Track failed purchase
track_recurring_charge_success()	Track subscription charge
track_recurring_charge_failed()	Track failed charge
identify()	Identify user with properties

track() - Custom Events

mf.track(
    user_id: int,                 # Telegram user ID
    event_name: str,              # Event name (e.g., 'STORY_RESPONSE')
    properties: dict = None,      # Any key-value pairs
)

Configuration

mf = MetricsFirst(
    bot_id="your_bot_id",
    api_key="your_api_key",
    api_url="https://api.metricsfirst.com",  # Custom API URL
    batch_events=True,      # Batch events before sending
    batch_size=10,          # Events per batch
    batch_interval=5.0,     # Seconds between flushes
    debug=False,            # Enable debug logging
    timeout=10.0,           # HTTP timeout
)

License

MIT