API Reference

Core Classes

TRC20Monitor

The main monitoring class that orchestrates transaction tracking.

class TRC20Monitor:
    def __init__(
        self,
        database: DatabaseAdapter,
        notifier: NotificationAdapter,
        max_retries: int = 3,
        retry_delay: float = 1.0
    )

Parameters:

database: Database adapter for storing transactions
notifier: Notification adapter for alerts
max_retries: Maximum number of retry attempts for failed requests
retry_delay: Delay in seconds between retries

Methods:

monitor_address

async def monitor_address(
    self,
    address: str,
    contract: str,
    direction: str = "in",
    interval: int = 30,
    min_amount: float = 0,
    duration: Optional[int] = None
) -> None

Monitor a Tron address for TRC20 transactions.

Parameters:

address: Tron address to monitor
contract: TRC20 contract address
direction: Transaction direction (“in”, “out”, or “both”)
interval: Check interval in seconds
min_amount: Minimum transaction amount to process
duration: Total monitoring duration in seconds (None for infinite)

Models

Transaction

Represents a TRC20 transaction.

@dataclass
class Transaction:
    tx_hash: str
    from_address: str
    to_address: str
    amount: Decimal
    contract: str
    timestamp: int
    block_number: int
    direction: Optional[str] = None

Attributes:

tx_hash: Transaction hash
from_address: Sender address
to_address: Recipient address
amount: Transaction amount (with decimals)
contract: TRC20 contract address
timestamp: Unix timestamp
block_number: Block number
direction: Transaction direction relative to monitored address

Adapters

DatabaseAdapter

Abstract base class for database implementations.

class DatabaseAdapter(ABC):
    @abstractmethod
    async def save_transaction(self, transaction: Transaction) -> None:
        """Save a transaction to the database."""
        pass
    
    @abstractmethod
    async def get_latest_transaction(
        self, address: str, contract: str, direction: str
    ) -> Optional[Transaction]:
        """Get the latest transaction for an address."""
        pass
    
    @abstractmethod
    async def transaction_exists(self, tx_hash: str) -> bool:
        """Check if a transaction exists."""
        pass

NotificationAdapter

Abstract base class for notification implementations.

class NotificationAdapter(ABC):
    @abstractmethod
    async def notify(self, transaction: Transaction) -> None:
        """Send notification for a transaction."""
        pass

Built-in Implementations

Database Adapters

MemoryDB

In-memory database implementation.

class MemoryDB(DatabaseAdapter):
    def __init__(self)

Usage:

db = MemoryDB()

SQLiteDB

SQLite database implementation.

class SQLiteDB(DatabaseAdapter):
    def __init__(self, db_path: str = "transactions.db")

Parameters:

db_path: Path to SQLite database file

Usage:

db = SQLiteDB("my_transactions.db")

Notification Adapters

ConsoleNotifier

Prints notifications to console.

class ConsoleNotifier(NotificationAdapter):
    def __init__(self, format_string: Optional[str] = None)

Parameters:

format_string: Custom format string for output

Usage:

notifier = ConsoleNotifier()

FileNotifier

Writes notifications to a file.

class FileNotifier(NotificationAdapter):
    def __init__(self, file_path: str, append: bool = True)

Parameters:

file_path: Path to output file
append: Whether to append or overwrite

Usage:

notifier = FileNotifier("transactions.log")

WebhookNotifier

Sends notifications to a webhook URL.

class WebhookNotifier(NotificationAdapter):
    def __init__(
        self,
        url: str,
        headers: Optional[Dict[str, str]] = None,
        timeout: int = 30
    )

Parameters:

url: Webhook endpoint URL
headers: Optional HTTP headers
timeout: Request timeout in seconds

Usage:

notifier = WebhookNotifier(
    url="https://webhook.site/your-endpoint",
    headers={"Authorization": "Bearer token"}
)

CompositeNotifier

Combines multiple notifiers.

class CompositeNotifier(NotificationAdapter):
    def __init__(self, notifiers: List[NotificationAdapter])

Parameters:

notifiers: List of notification adapters

Usage:

notifier = CompositeNotifier([
    ConsoleNotifier(),
    FileNotifier("log.txt"),
    WebhookNotifier("https://webhook.site/endpoint")
])

Exceptions

MonitorError

Base exception for monitoring errors.

class MonitorError(Exception):
    """Base exception for monitor errors."""

NetworkError

Network-related errors.

class NetworkError(MonitorError):
    """Network communication errors."""

ValidationError

Input validation errors.

class ValidationError(MonitorError):
    """Input validation errors."""

DatabaseError

Database operation errors.

class DatabaseError(MonitorError):
    """Database operation errors."""

NotificationError

Notification sending errors.

class NotificationError(MonitorError):
    """Notification sending errors."""

Utilities

Address Validation

from trc20_monitor.utils import is_valid_tron_address

if is_valid_tron_address("TYourAddress"):
    print("Valid address")

Amount Conversion

from trc20_monitor.utils import format_amount, parse_amount

# Format for display
formatted = format_amount(1000000, decimals=6)  # "1.0"

# Parse from string
amount = parse_amount("1.5", decimals=6)  # 1500000

Retry Decorator

from trc20_monitor.utils import retry

@retry(max_attempts=3, delay=1.0)
async def unstable_operation():
    # Your code here
    pass

Custom Adapters

Implementing a Custom Database

from trc20_monitor import DatabaseAdapter, Transaction
from typing import Optional

class PostgreSQLDB(DatabaseAdapter):
    def __init__(self, connection_string: str):
        self.conn_string = connection_string
        # Initialize connection
    
    async def save_transaction(self, transaction: Transaction) -> None:
        # Save to PostgreSQL
        query = """
            INSERT INTO transactions (tx_hash, from_address, to_address, ...)
            VALUES ($1, $2, $3, ...)
        """
        # Execute query
    
    async def get_latest_transaction(
        self, address: str, contract: str, direction: str
    ) -> Optional[Transaction]:
        # Query PostgreSQL
        query = """
            SELECT * FROM transactions
            WHERE address = $1 AND contract = $2 AND direction = $3
            ORDER BY timestamp DESC
            LIMIT 1
        """
        # Return result
    
    async def transaction_exists(self, tx_hash: str) -> bool:
        # Check existence
        query = "SELECT EXISTS(SELECT 1 FROM transactions WHERE tx_hash = $1)"
        # Return result

Implementing a Custom Notifier

from trc20_monitor import NotificationAdapter, Transaction
import smtplib

class EmailNotifier(NotificationAdapter):
    def __init__(self, smtp_host: str, smtp_port: int, 
                 username: str, password: str, to_email: str):
        self.smtp_host = smtp_host
        self.smtp_port = smtp_port
        self.username = username
        self.password = password
        self.to_email = to_email
    
    async def notify(self, transaction: Transaction) -> None:
        subject = f"New TRC20 Transaction: {transaction.amount}"
        body = f"""
        Transaction Details:
        - Hash: {transaction.tx_hash}
        - From: {transaction.from_address}
        - To: {transaction.to_address}
        - Amount: {transaction.amount}
        - Block: {transaction.block_number}
        """
        
        # Send email using SMTP
        with smtplib.SMTP(self.smtp_host, self.smtp_port) as server:
            server.starttls()
            server.login(self.username, self.password)
            message = f"Subject: {subject}\\n\\n{body}"
            server.sendmail(self.username, self.to_email, message)

Type Hints

The library is fully typed and works with mypy:

from typing import Optional, List
from trc20_monitor import TRC20Monitor, Transaction

async def process_transactions(
    monitor: TRC20Monitor,
    addresses: List[str],
    min_amount: Optional[float] = None
) -> List[Transaction]:
    # Your typed code here
    pass