'
}
try:
# Load active simulations
active_path = os.path.join(DATA_DIR, "simulations", "active.json")
if os.path.exists(active_path):
with open(active_path) as f:
active = json.load(f)
positions = active.get('positions', [])
data['active_simulations'] = len(positions)
data['total_bankroll'] = active.get('bankroll_used', 0)
# Calculate P&L
total_pnl = sum(pos.get('unrealized_pnl', 0) for pos in positions)
data['total_pnl'] = total_pnl
data['total_pnl_class'] = 'positive' if total_pnl >= 0 else 'negative'
if data['total_bankroll'] > 0:
data['total_pnl_pct'] = (total_pnl / data['total_bankroll']) * 100
# Render active positions
if positions:
pos_html = ""
for pos in positions:
pos_html += f"""
{pos.get('asset', 'Unknown')[:50]}
${pos.get('unrealized_pnl', 0):,.0f}
"""
data['active_positions'] = pos_html
# Count investigations today
inv_pattern = os.path.join(DATA_DIR, "investigations", "inv-*.json")
today = datetime.now().strftime('%Y%m%d')
today_invs = [f for f in glob.glob(inv_pattern) if today in os.path.basename(f)]
data['investigated_today'] = len(today_invs)
# Recent activity from investigations
if today_invs:
activity_html = ""
for inv_file in today_invs[-5:]: # Last 5
with open(inv_file) as f:
inv = json.load(f)
verdict = inv.get('investigation', {}).get('verdict', 'Unknown')
author = inv.get('source_post', {}).get('author', 'Unknown')
activity_html += f"""
TodayInvestigated {author}: {verdict}
"""
data['recent_activity'] = activity_html
except Exception as e:
print(f"Dashboard data error: {e}")
return data
def get_latest_posts(self):
"""Get latest scraped posts with triage data"""
posts = []
try:
# Find latest x-feed directory
x_feed_pattern = os.path.join("../../data/x-feed", "20*")
x_feed_dirs = sorted(glob.glob(x_feed_pattern))
if x_feed_dirs:
latest_dir = x_feed_dirs[-1]
posts_file = os.path.join(latest_dir, "posts.json")
triage_file = os.path.join(latest_dir, "triage.json")
# Load posts
if os.path.exists(posts_file):
with open(posts_file) as f:
posts_data = json.load(f)
posts = posts_data.get('posts', [])
# Load triage data and merge
if os.path.exists(triage_file):
with open(triage_file) as f:
triage = json.load(f)
# Merge triage data with posts
for post in posts:
post_id = post.get('id')
for category, triaged_posts in triage.items():
if category == 'investigation_queue':
continue
for tpost in triaged_posts:
if tpost.get('id') == post_id:
post['triage'] = {
'category': category,
'priority': tpost.get('priority'),
'reason': tpost.get('reason')
}
break
except Exception as e:
print(f"Posts loading error: {e}")
return posts[:50] # Limit to last 50
def get_investigations(self):
"""Get all investigation reports"""
investigations = []
try:
inv_pattern = os.path.join(DATA_DIR, "investigations", "inv-*.json")
for inv_file in sorted(glob.glob(inv_pattern), reverse=True):
with open(inv_file) as f:
inv = json.load(f)
investigations.append(inv)
except Exception as e:
print(f"Investigations loading error: {e}")
return investigations
def get_simulation_data(self):
"""Get simulation data (active + history)"""
data = {'active': {}, 'history': []}
try:
# Load active positions
active_path = os.path.join(DATA_DIR, "simulations", "active.json")
if os.path.exists(active_path):
with open(active_path) as f:
data['active'] = json.load(f)
# Load history if it exists
history_path = os.path.join(DATA_DIR, "simulations", "history.json")
if os.path.exists(history_path):
with open(history_path) as f:
data['history'] = json.load(f)
except Exception as e:
print(f"Simulation data error: {e}")
return data
def get_status_data(self):
"""Get pipeline status information"""
status = {
'chrome': {'status': 'Unknown', 'class': 'unknown', 'detail': 'Checking...'},
'last_run': {'time': 'Unknown', 'detail': 'No recent runs found'},
'next_run': {'time': 'Unknown', 'detail': 'Check schedule in config.json'},
'health': []
}
try:
# Check Chrome debug port
import urllib.request
try:
urllib.request.urlopen('http://127.0.0.1:9222/json', timeout=2)
status['chrome'] = {
'status': 'Running',
'class': 'healthy',
'detail': 'Debug port 9222 accessible'
}
except:
status['chrome'] = {
'status': 'Down',
'class': 'error',
'detail': 'Port 9222 not accessible'
}
# Check for recent pipeline runs
x_feed_pattern = os.path.join("../../data/x-feed", "20*")
x_feed_dirs = sorted(glob.glob(x_feed_pattern))
if x_feed_dirs:
latest = os.path.basename(x_feed_dirs[-1])
# Parse timestamp from dirname (format: 20260207-234451)
try:
dt = datetime.strptime(latest, '%Y%m%d-%H%M%S')
status['last_run'] = {
'time': dt.strftime('%Y-%m-%d %H:%M:%S'),
'detail': f"Scraped data in {latest}"
}
except:
status['last_run'] = {
'time': latest,
'detail': 'Found recent scrape data'
}
except Exception as e:
print(f"Status check error: {e}")
return status
# Rendering methods
def render_posts(self, posts):
"""Render feed posts with triage status"""
if not posts:
return '
No posts found
'
html = ""
for post in posts:
triage = post.get('triage', {})
category = triage.get('category', 'unknown')
priority = triage.get('priority', 0)
# Color coding based on triage category
if category == 'high_value':
status_class = 'high-value'
status_text = f'High Value (Priority: {priority})'
elif category == 'worth_investigating':
status_class = 'investigate'
status_text = f'Worth Investigating (Priority: {priority})'
elif category == 'dismissed':
status_class = 'dismissed'
status_text = 'Dismissed'
else:
status_class = 'unknown'
status_text = 'Not Triaged'
html += f"""
{post.get('author', 'Unknown')}
{status_text}
{post.get('text', '')[:200]}...
"""
return html
def render_investigations(self, investigations):
"""Render investigation reports"""
if not investigations:
return '
No investigations found
'
html = ""
for inv in investigations:
investigation = inv.get('investigation', {})
verdict = investigation.get('verdict', 'Unknown')
risk_score = investigation.get('risk_assessment', {}).get('score', 0)
source = inv.get('source_post', {})
verdict_class = 'verified' if 'VERIFIED' in verdict else 'failed'
html += f"""
{source.get('author', 'Unknown')}
{verdict}
{source.get('claim', 'No claim')}
Risk Score: {risk_score}/10
"""
return html
def render_active_positions(self, active_data):
"""Render active trading positions"""
positions = active_data.get('positions', [])
if not positions:
return '
No active positions
'
html = ""
for pos in positions:
pnl = pos.get('unrealized_pnl', 0)
pnl_class = 'positive' if pnl >= 0 else 'negative'
html += f"""
"""
return html
def render_performance_metrics(self, sim_data):
"""Render performance metrics"""
active = sim_data.get('active', {})
positions = active.get('positions', [])
if not positions:
return '
No performance data
'
total_pnl = sum(pos.get('unrealized_pnl', 0) for pos in positions)
bankroll = active.get('bankroll_used', 1)
pnl_pct = (total_pnl / bankroll) * 100 if bankroll > 0 else 0
return f"""
Total P&L
${total_pnl:,.0f}
Return %
{pnl_pct:+.1f}%
Bankroll Used
${bankroll:,.0f}
"""
def render_trade_history(self, history):
"""Render trade history"""
if not history:
return '
No trade history
'
html = ""
for trade in history[-10:]: # Last 10 trades
pnl = trade.get('realized_pnl', 0)
pnl_class = 'positive' if pnl >= 0 else 'negative'
html += f"""