PnL algorithm using getTransactionsForAddress

README.md

sol-pnl — single-file edition

Compute SOL profit-and-loss for a Solana wallet via the Helius API.
Everything lives in one file: lib.rs. No project setup required. IN CASE YOU NEED FULL REPO: https://github.com/MattMsh/low-latency-pnl-helius

Run (one command)

# Install runner once
cargo install rust-script

# Run
rust-script lib.rs <WALLET_ADDRESS> --api-key <HELIUS_API_KEY>

# Or put your key in .env and omit the flag
HELIUS_API_KEY=xxx rust-script lib.rs <WALLET_ADDRESS>

rust-script reads the embedded [dependencies] block at the top of the file, resolves them via crates.io, and runs — no Cargo.toml needed.

Flags

Flag	Env	Default	Description
<address>	—	required	Wallet address (base58)
--api-key	HELIUS_API_KEY	required	Helius API key
--concurrency	—	10	Parallel RPC requests
--verbose	—	false	Full JSON output

Output

Address:  <wallet>
PnL:      1.234567890 SOL (1234567890 lamports)
Txns:     42
First tx: 1700000000
Last tx:  1710000000
Total:    5032 ms

Phase timings via RUST_LOG=info:

phase=discovery  sig_count=1234  elapsed_ms=820
phase=fetch      tx_count=1234   elapsed_ms=4210
phase=compute    elapsed_us=3

Algorithm (3 passes)

Pass	API mode	Page limit	Purpose
1 — Discovery	Signatures	1 000	Collect all blockTimes cheaply
2 — Fetch	Full / JsonParsed	100, parallel	Full tx data in time-range chunks
3 — Compute	CPU	—	Σ (post_balance − pre_balance)

Chunking: signatures are split into N non-overlapping blockTime windows (N = --concurrency) so all chunks fetch in parallel without duplicates.

Make it executable (optional)

chmod +x lib.rs
./lib.rs <WALLET_ADDRESS> --api-key <KEY>

Works because the shebang #!/usr/bin/env rust-script is already at the top.

main.rs

#!/usr/bin/env rust-script
//! ```cargo
//! [dependencies]
//! helius      = "1.0.1"
//! tokio       = { version = "1", features = ["full"] }
//! serde       = { version = "1", features = ["derive"] }
//! serde_json  = "1"
//! thiserror   = "2"
//! tracing     = "0.1"
//! tracing-subscriber = "0.3"
//! dotenvy     = "0.15"
//! ```

use std::sync::Arc;

use helius::Helius;
use helius::types::{
    BlockTimeFilter, Cluster, GetTransactionsFilters, GetTransactionsForAddressOptions, SortOrder,
    TransactionDetails, TransactionStatusFilter, UiTransactionEncoding,
};
use tokio::sync::Semaphore;

// ── Errors ────────────────────────────────────────────────────────────────────

#[derive(Debug, thiserror::Error)]
pub enum PnlError {
    #[error("Helius API error: {0}")]
    Helius(#[from] helius::error::HeliusError),
    #[error("Task join error: {0}")]
    Join(#[from] tokio::task::JoinError),
    #[error("No transactions found for this address")]
    NoTransactions,
    #[error("Semaphore error: {0}")]
    Semaphore(#[from] tokio::sync::AcquireError),
    #[error("JSON parse error: {0}")]
    Json(#[from] serde_json::Error),
}

pub type Result<T> = std::result::Result<T, PnlError>;

// ── Config ────────────────────────────────────────────────────────────────────

pub struct PnlConfig {
    pub api_key: String,
    pub address: String,
    pub cluster: Cluster,
    pub max_concurrency: usize,
}

// ── Discovery (Pass 1) ────────────────────────────────────────────────────────

struct SigEntry {
    block_time: i64,
}

async fn fetch_all_signatures(helius: &Helius, address: &str) -> Result<Vec<SigEntry>> {
    let rpc = helius.rpc();
    let addr = address.to_string();
    let mut entries: Vec<SigEntry> = Vec::new();
    let mut pagination_token: Option<String> = None;

    loop {
        let opts = GetTransactionsForAddressOptions {
            transaction_details: Some(TransactionDetails::Signatures),
            sort_order: Some(SortOrder::Asc),
            limit: Some(1000),
            filters: Some(GetTransactionsFilters {
                status: Some(TransactionStatusFilter::Succeeded),
                ..Default::default()
            }),
            pagination_token: pagination_token.take(),
            ..Default::default()
        };

        let resp = rpc.get_transactions_for_address(addr.clone(), opts).await?;

        for val in &resp.data {
            if let Some(bt) = val["blockTime"].as_i64() {
                entries.push(SigEntry { block_time: bt });
            }
        }

        match resp.pagination_token {
            Some(token) => pagination_token = Some(token),
            None => break,
        }
    }

    entries.sort_by_key(|e| e.block_time);
    Ok(entries)
}

// ── Chunking ──────────────────────────────────────────────────────────────────

struct FetchChunk {
    start: i64,
    end: i64,
}

/// Split signatures into `n` balanced chunks by tx count.
fn generate_chunks(sigs: &[SigEntry], n: usize) -> Vec<FetchChunk> {
    if sigs.is_empty() || n == 0 {
        return Vec::new();
    }

    let n = n.min(sigs.len());
    let chunk_size = sigs.len() / n;
    let remainder = sigs.len() % n;

    let mut chunks = Vec::with_capacity(n);
    let mut offset = 0;

    for i in 0..n {
        let size = chunk_size + if i < remainder { 1 } else { 0 };
        let start = sigs[offset].block_time;
        let end = sigs[offset + size - 1].block_time;
        chunks.push(FetchChunk { start, end });
        offset += size;
    }

    if let Some(first) = chunks.first_mut() {
        first.start -= 1;
    }
    if let Some(last) = chunks.last_mut() {
        last.end += 1;
    }

    for i in 1..chunks.len() {
        let boundary = chunks[i].start;
        chunks[i - 1].end = boundary - 1;
    }

    chunks
}

// ── Fetcher (Pass 2) ──────────────────────────────────────────────────────────

#[derive(Debug, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
struct FullTransactionItem {
    slot: u64,
    block_time: Option<i64>,
    transaction_index: Option<u64>,
    transaction: FullTransaction,
    meta: FullTransactionMeta,
}

#[derive(Debug, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
struct FullTransaction {
    signatures: Vec<String>,
    message: FullTransactionMessage,
}

#[derive(Debug, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
struct FullTransactionMessage {
    account_keys: Vec<serde_json::Value>,
}

#[derive(Debug, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
struct FullTransactionMeta {
    pre_balances: Vec<u64>,
    post_balances: Vec<u64>,
}

async fn fetch_all(
    helius: &Helius,
    address: &str,
    chunks: Vec<FetchChunk>,
    max_concurrency: usize,
) -> Result<Vec<FullTransactionItem>> {
    let sem = Arc::new(Semaphore::new(max_concurrency));
    let addr = address.to_string();
    let mut handles = Vec::with_capacity(chunks.len());

    for chunk in chunks {
        let sem = sem.clone();
        let rpc = helius.rpc();
        let addr = addr.clone();

        handles.push(tokio::spawn(async move {
            let _permit = sem.acquire_owned().await?;
            let mut items: Vec<FullTransactionItem> = Vec::new();
            let mut pagination_token: Option<String> = None;

            loop {
                let opts = GetTransactionsForAddressOptions {
                    transaction_details: Some(TransactionDetails::Full),
                    sort_order: Some(SortOrder::Asc),
                    limit: Some(100),
                    encoding: Some(UiTransactionEncoding::JsonParsed),
                    max_supported_transaction_version: Some(0),
                    filters: Some(GetTransactionsFilters {
                        block_time: Some(BlockTimeFilter {
                            gte: Some(chunk.start),
                            lte: Some(chunk.end),
                            ..Default::default()
                        }),
                        status: Some(TransactionStatusFilter::Succeeded),
                        ..Default::default()
                    }),
                    pagination_token: pagination_token.take(),
                    ..Default::default()
                };

                let resp = rpc.get_transactions_for_address(addr.clone(), opts).await?;
                let next_token = resp.pagination_token;

                for val in resp.data {
                    match serde_json::from_value::<FullTransactionItem>(val) {
                        Ok(item) => items.push(item),
                        Err(e) => tracing::warn!("Skipping unparseable transaction: {}", e),
                    }
                }

                match next_token {
                    Some(token) => pagination_token = Some(token),
                    None => break,
                }
            }

            Ok::<Vec<FullTransactionItem>, PnlError>(items)
        }));
    }

    let mut all: Vec<FullTransactionItem> = Vec::new();
    for handle in handles {
        all.extend(handle.await??);
    }

    all.sort_by_key(|item| (item.slot, item.transaction_index.unwrap_or(0)));
    Ok(all)
}

// ── PnL computation (Pass 3) ──────────────────────────────────────────────────

#[derive(Debug, serde::Serialize)]
pub struct PnlEntry {
    pub signature: String,
    pub block_time: i64,
    pub sol_change_lamports: i64,
    pub running_balance_lamports: i64,
}

#[derive(Debug, serde::Serialize)]
pub struct PnlReport {
    pub address: String,
    pub total_pnl_lamports: i64,
    pub total_pnl_sol: f64,
    pub tx_count: usize,
    pub first_tx_time: Option<i64>,
    pub last_tx_time: Option<i64>,
    pub entries: Vec<PnlEntry>,
}

fn extract_pubkey(key: &serde_json::Value) -> Option<&str> {
    if let Some(s) = key.as_str() {
        Some(s)
    } else {
        key.get("pubkey").and_then(|v| v.as_str())
    }
}

fn compute(address: &str, transactions: &[FullTransactionItem]) -> PnlReport {
    let mut entries: Vec<PnlEntry> = Vec::new();
    let mut running_balance: i64 = 0;

    for tx in transactions {
        let account_keys = &tx.transaction.message.account_keys;

        let idx = account_keys.iter().enumerate().find_map(|(i, key)| {
            if extract_pubkey(key) == Some(address) { Some(i) } else { None }
        });

        let idx = match idx {
            Some(i) => i,
            None => continue,
        };

        let pre = tx.meta.pre_balances.get(idx).copied().unwrap_or(0) as i64;
        let post = tx.meta.post_balances.get(idx).copied().unwrap_or(0) as i64;
        let delta = post - pre;
        running_balance += delta;

        entries.push(PnlEntry {
            signature: tx.transaction.signatures.first().cloned().unwrap_or_default(),
            block_time: tx.block_time.unwrap_or(0),
            sol_change_lamports: delta,
            running_balance_lamports: running_balance,
        });
    }

    let total_pnl_lamports = running_balance;
    PnlReport {
        address: address.to_string(),
        total_pnl_lamports,
        total_pnl_sol: total_pnl_lamports as f64 / 1_000_000_000.0,
        tx_count: entries.len(),
        first_tx_time: entries.first().map(|e| e.block_time),
        last_tx_time: entries.last().map(|e| e.block_time),
        entries,
    }
}

// ── Public API ────────────────────────────────────────────────────────────────

pub async fn compute_pnl(config: PnlConfig) -> Result<PnlReport> {
    let helius = Helius::new(&config.api_key, config.cluster)?;

    let t1 = std::time::Instant::now();
    let sigs = fetch_all_signatures(&helius, &config.address).await?;
    tracing::info!(phase = "discovery", sig_count = sigs.len(), elapsed_ms = t1.elapsed().as_millis());
    if sigs.is_empty() {
        return Err(PnlError::NoTransactions);
    }

    let chunks = generate_chunks(&sigs, config.max_concurrency);

    let t2 = std::time::Instant::now();
    let txs = fetch_all(&helius, &config.address, chunks, config.max_concurrency).await?;
    tracing::info!(phase = "fetch", tx_count = txs.len(), elapsed_ms = t2.elapsed().as_millis());

    let t3 = std::time::Instant::now();
    let report = compute(&config.address, &txs);
    tracing::info!(phase = "compute", elapsed_us = t3.elapsed().as_micros());

    Ok(report)
}

// ── Entry point ───────────────────────────────────────────────────────────────

#[tokio::main]
async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
    dotenvy::dotenv().ok();
    tracing_subscriber::fmt::init();

    let mut args = std::env::args().skip(1);
    let address = args.next().expect("Usage: lib.rs <WALLET_ADDRESS> [--api-key <KEY>] [--concurrency <N>] [--verbose]");

    let mut api_key = std::env::var("HELIUS_API_KEY").unwrap_or_default();
    let mut concurrency: usize = 10;
    let mut verbose = false;

    while let Some(flag) = args.next() {
        match flag.as_str() {
            "--api-key"     => api_key     = args.next().expect("--api-key requires a value"),
            "--concurrency" => concurrency = args.next().expect("--concurrency requires a value").parse()?,
            "--verbose"     => verbose     = true,
            other           => eprintln!("Unknown flag: {other}"),
        }
    }

    if api_key.is_empty() {
        eprintln!("Error: HELIUS_API_KEY env var or --api-key flag required");
        std::process::exit(1);
    }

    let t_total = std::time::Instant::now();
    let report = compute_pnl(PnlConfig {
        api_key,
        address,
        cluster: Cluster::MainnetBeta,
        max_concurrency: concurrency,
    }).await?;
    let total_ms = t_total.elapsed().as_millis();

    if verbose {
        println!("{}", serde_json::to_string_pretty(&report)?);
    } else {
        println!("Address:  {}", report.address);
        println!("PnL:      {:.9} SOL ({} lamports)", report.total_pnl_sol, report.total_pnl_lamports);
        println!("Txns:     {}", report.tx_count);
        if let Some(t) = report.first_tx_time { println!("First tx: {t}"); }
        if let Some(t) = report.last_tx_time  { println!("Last tx:  {t}"); }
        println!("Total:    {total_ms} ms");
    }

    Ok(())
}