quil-stats.py

#!/usr/bin/env python3
import sys
import subprocess
import argparse
from datetime import datetime
from pathlib import Path


def parse_cli_arguments():
    parser = argparse.ArgumentParser(description="Calculate statistics of Quil mining.")
    parser.add_argument("-p", "--client-path", type=str, default="../client/qclient", help="Path to the qClient.")
    parser.add_argument("-c", "--config-path", type=str, default="./.config", help="Path to the .config directory.")
    parser.add_argument("-s", "--start-frame", type=int, help="Specify the starting frame number.")
    parser.add_argument("-e", "--end-frame", type=int, help="Specify the ending frame number. Leave empty to use latest frame.")
    parser.add_argument("-n", "--latest-frames", type=int, help="Specify the number of latest frames to calculate statistics.")
    parser.add_argument("-w", "--workers", type=int, help="Specify the number of workers.")
    parser.add_argument("-v", "--verbose", action="store_true", help="Enable verbose output.")
    return parser.parse_args()


def get_coins(path: Path, config: Path):
    if not path.exists():
        sys.exit(f"Path {path} does not exists!")
    try:
        result = subprocess.run(
            [path, "--signature-check=false", "--config", config, "--public-rpc", "token", "coins", "metadata"],
            capture_output=True, text=True, check=True
        )
        return result.stdout.splitlines()
    except subprocess.CalledProcessError as e:
        print("Error fetching coins:", e)
        return []


def parse_coins(coins):
    parsed_data = []
    for line in coins:
        try:
            coin_value_str = line.split()[0].strip()
            coin_value = float(coin_value_str)

            address_start = line.find("Coin") + len("Coin ")
            address_end = line.find(")", address_start)
            address = line[address_start:address_end].strip()

            frame_part = line.split("Frame")[-1].split(",")[0].strip()
            frame = int(frame_part)

            timestamp_part = line.split("Timestamp")[-1].strip()
            timestamp = datetime.fromisoformat(timestamp_part.replace("Z", "+00:00"))

            parsed_data.append({
                "value": coin_value,
                "address": address,
                "frame": frame,
                "timestamp": timestamp,
                "line": line
            })
        except (ValueError, IndexError) as e:
            if "Signature" not in line and "gRPC" not in line and "Public RPC" not in line:
                print(f"Error parsing line: {line}, error: {e}")
            continue

    return parsed_data


def filter_frames(parsed_data, start_frame, end_frame, latest_frames):
    if start_frame is None and end_frame is None and latest_frames is None:
        return start_frame, end_frame, parsed_data

    parsed_data = sorted(parsed_data, key=lambda x: x["frame"])
    if latest_frames is not None:
        start_frame = parsed_data[-1]["frame"] - latest_frames
    if end_frame is None:
        end_frame = parsed_data[-1]["frame"]

    parsed_data = [
        entry for entry in parsed_data
        if (start_frame is None or entry["frame"] >= start_frame) and
            (end_frame is None or entry["frame"] <= end_frame)
    ]
    return start_frame, end_frame, parsed_data


def calculate_statistics(filtered_data, start_frame, end_frame, workers):
    landed_frames = len(filtered_data)
    reward = sum(x['value'] for x in filtered_data)
    reward_rate = reward / (filtered_data[-1]["timestamp"] - filtered_data[0]["timestamp"]).total_seconds() * 3600 * 24
    reward_rate_per_worker = reward_rate / workers if workers else None
    reward_per_worker_frame = reward / workers / landed_frames if workers else None
    total_frames = end_frame - start_frame + 1
    if total_frames > 0:
        landing_rate = landed_frames / total_frames * 100
    else:
        landing_rate = 0.0
    return {
        "landed_frames": landed_frames, 
        "total_frames": total_frames, 
        "landing_rate": landing_rate, 
        "reward": reward, 
        "reward_rate": reward_rate,
        "reward_rate_per_worker": reward_rate_per_worker,
        "reward_per_worker_frame": reward_per_worker_frame
    }


def main():
    args = parse_cli_arguments()
    coins = get_coins(Path(args.client_path), Path(args.config_path))
    if not coins:
        sys.exit("No coins retrieved.")

    parsed_data = parse_coins(coins)
    start_frame, end_frame, filtered_data = filter_frames(parsed_data, args.start_frame, args.end_frame, args.latest_frames)
    stats = calculate_statistics(filtered_data, start_frame, end_frame, args.workers)

    if args.verbose:
        print("\nFiltered Landed Frames:")
        for entry in filtered_data:
            print(entry["line"])

    print(f"Statistics for frames {start_frame} to {end_frame}:")
    print(f"  Total Frames: {stats['total_frames']}")
    print(f"  Landed Frames: {stats['landed_frames']}")
    print(f"  Landing Rate: {stats['landing_rate']:.2f} %")
    print(f"  Reward: {stats['reward']:.4f} QUIL")
    print(f"  Reward Rate: {stats['reward_rate']:.4f} QUIL/day")
    if stats['reward_rate_per_worker']:
        print(f"  Workers: {args.workers}")
        print(f"  Reward Rate per Worker: {stats['reward_rate_per_worker']:.4f} QUIL/day/worker")
        print(f"  Reward per Worker Landed Frame: {stats['reward_per_worker_frame']:.5f} QUIL/worker/frame")


if __name__ == "__main__":
    main()


# ===== Tests =====
COINS = """0.033733312000 QUIL (Coin 0x10adef098048407776a810ae79468e37ebaa25ee5bf1ba196f6b4b5bb1123456) Frame 63601, Timestamp 2024-11-27T05:35:29Z
0.030809036000 QUIL (Coin 0x10adef098048407776a810ae79468e37ebaa25ee5bf1ba196f6b4b5bb1123456) Frame 63600, Timestamp 2024-11-27T05:33:17Z
0.230336389500 QUIL (Coin 0x10adef098048407776a810ae79468e37ebaa25ee5bf1ba196f6b4b5bb1123456) Frame 63543, Timestamp 2024-11-27T03:24:51Z
0.224817039000 QUIL (Coin 0x10adef098048407776a810ae79468e37ebaa25ee5bf1ba196f6b4b5bb1123456) Frame 63517, Timestamp 2024-11-27T02:28:08Z
0.223805187000 QUIL (Coin 0x10adef098048407776a810ae79468e37ebaa25ee5bf1ba196f6b4b5bb1123456) Frame 63516, Timestamp 2024-11-27T02:25:56Z
0.265759245000 QUIL (Coin 0x10adef098048407776a810ae79468e37ebaa25ee5bf1ba196f6b4b5bb1123456) Frame 63511, Timestamp 2024-11-27T02:14:57Z
0.219922506000 QUIL (Coin 0x10adef098048407776a810ae79468e37ebaa25ee5bf1ba196f6b4b5bb1123456) Frame 63479, Timestamp 2024-11-27T01:03:06Z
0.203279166000 QUIL (Coin 0x10adef098048407776a810ae79468e37ebaa25ee5bf1ba196f6b4b5bb1123456) Frame 63478, Timestamp 2024-11-27T01:00:52Z
0.249665868000 QUIL (Coin 0x10adef098048407776a810ae79468e37ebaa25ee5bf1ba196f6b4b5bb1123456) Frame 63477, Timestamp 2024-11-27T00:58:33+00:00"""


def test_calculate_statistics():
    workers = 3
    parsed_data = parse_coins(COINS.splitlines())
    start_frame, end_frame, filtered_data = filter_frames(parsed_data, 63477, None, None)
    stats = calculate_statistics(filtered_data, start_frame, end_frame, workers)
    
    assert stats["landed_frames"] == 9
    assert stats["total_frames"] == 63601 - 63477 + 1
    assert stats["landing_rate"] == 9 / (63601 - 63477 + 1) * 100
    assert stats["reward"] == 1.6821277485
    assert stats["reward_rate"] == stats["reward"] / 16616 * 3600 * 24
    assert stats["reward_per_worker_frame"] == stats["reward"] / workers / 9