Prathamesh Musale
652c69cbee
- Refactor lockdrop calculations and other helper code from the simulation notebook to a python module - Add a notebook for experimenting with lockdrop calculations - Add widget buttons to run calculations and export results - Add a script a to setup and run the notebook Reviewed-on: #2 Co-authored-by: Prathamesh Musale <prathamesh.musale0@gmail.com> Co-committed-by: Prathamesh Musale <prathamesh.musale0@gmail.com>
130 lines
4.4 KiB
Python
130 lines
4.4 KiB
Python
"""
|
|
Simulation-specific functions for processing zenith-stack generated data.
|
|
|
|
This module contains functions specifically for processing simulation data
|
|
generated by the zenith-stack project, including watcher events and participant data.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import urbitob
|
|
from decimal import Decimal
|
|
from collections import defaultdict
|
|
|
|
from .constants import NUM_GALAXIES
|
|
from .calculations import (
|
|
calculate_dynamic_allocations, calculate_bonus_pools,
|
|
calculate_final_allocations, generate_test_output
|
|
)
|
|
from .display import print_analysis_tables
|
|
|
|
|
|
def simulation_load_watcher_events(file_path):
|
|
"""Load and parse PointLockedEvent events from JSON file (simulation-specific)."""
|
|
with open(file_path, 'r') as f:
|
|
data = json.load(f)
|
|
|
|
# Filter for only PointLockedEvent events during loading
|
|
point_locked_events = [
|
|
event_data for event_data in data['data']['eventsInRange']
|
|
if event_data['event']['__typename'] == 'PointLockedEvent'
|
|
]
|
|
|
|
return point_locked_events
|
|
|
|
|
|
def simulation_analyze_lockdrop_events(events):
|
|
"""Analyze PointLockedEvent events and return participation statistics (simulation-specific)."""
|
|
# Initialize counters
|
|
lock_duration_counts = {
|
|
'star': defaultdict(int),
|
|
'galaxy': defaultdict(int)
|
|
}
|
|
|
|
# Process events (already filtered to PointLockedEvent only)
|
|
for event_data in events:
|
|
point = event_data['event']['point']
|
|
lock_period = event_data['event']['lock_period']
|
|
|
|
# Determine if it's a galaxy or star
|
|
point_num = urbitob.patp_to_num(point)
|
|
point_type = "galaxy" if point_num < NUM_GALAXIES else "star"
|
|
|
|
# Count by lock period
|
|
lock_duration_counts[point_type][lock_period] += 1
|
|
|
|
# Extract counts for each year and point type
|
|
result = {}
|
|
for years in [1, 2, 3, 4, 5]:
|
|
result[f'stars_{years}_years'] = lock_duration_counts['star'][years]
|
|
result[f'galaxies_{years}_years'] = lock_duration_counts['galaxy'][years]
|
|
|
|
return result
|
|
|
|
|
|
def run_simulation_analysis(generated_dir=None):
|
|
"""Complete simulation analysis pipeline (simulation-specific)."""
|
|
# Determine generated directory
|
|
if generated_dir is None:
|
|
generated_dir = os.getenv('GENERATED_DIR', 'generated')
|
|
|
|
# Load and analyze watcher events
|
|
watcher_events_path = os.path.join(generated_dir, 'watcher-events.json')
|
|
|
|
print("=" * 80)
|
|
print("📁 WATCHER EVENTS DATA SUMMARY")
|
|
print("=" * 80)
|
|
|
|
try:
|
|
point_locked_events = simulation_load_watcher_events(watcher_events_path)
|
|
|
|
if point_locked_events:
|
|
total_events = len(point_locked_events)
|
|
print(f"✅ Successfully loaded {total_events} PointLockedEvent records from watcher file")
|
|
else:
|
|
print("⚠️ No PointLockedEvent events found")
|
|
return None
|
|
|
|
# Analyze events and convert to Decimal
|
|
lock_stats = simulation_analyze_lockdrop_events(point_locked_events)
|
|
participation_counts = {k: Decimal(v) for k, v in lock_stats.items()}
|
|
|
|
# Run calculations
|
|
allocation_data = calculate_dynamic_allocations(participation_counts)
|
|
bonus_data = calculate_bonus_pools(allocation_data)
|
|
final_data = calculate_final_allocations(allocation_data, bonus_data)
|
|
|
|
# Print comprehensive analysis
|
|
print_analysis_tables(allocation_data, bonus_data, final_data)
|
|
|
|
# Generate and save test output
|
|
test_output = generate_test_output(final_data)
|
|
allocations_output_file = 'lockdrop_allocations_notebook.json'
|
|
|
|
export_data = {
|
|
'metadata': {
|
|
'source': 'lockdrop-calculations-simulated.ipynb',
|
|
'participation_counts': lock_stats
|
|
},
|
|
'allocations': test_output
|
|
}
|
|
|
|
with open(allocations_output_file, 'w') as f:
|
|
json.dump(export_data, f, indent=2)
|
|
|
|
print("=" * 80)
|
|
print("💾 JSON OUTPUT GENERATED")
|
|
print(f"✅ Final allocations saved to: {allocations_output_file}")
|
|
print("=" * 80)
|
|
|
|
# Return data for visualization
|
|
return allocation_data, final_data
|
|
|
|
except FileNotFoundError:
|
|
print(f"❌ Watcher events file not found: {watcher_events_path}")
|
|
print(" Make sure GENERATED_DIR environment variable points to the correct directory")
|
|
return None
|
|
except Exception as e:
|
|
print(f"❌ Error during simulation analysis: {e}")
|
|
return None
|