236 lines
7.6 KiB
Python
Executable File
236 lines
7.6 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Enhanced BIGGFISH Daily Reporter
|
|
Redesigned for readability and actionable insights.
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Dict, List, Optional
|
|
|
|
# Load status data
|
|
def load_status() -> Dict:
|
|
status_file = Path("/opt/biggfish/src/data/krystie-status.json")
|
|
if not status_file.exists():
|
|
return {}
|
|
with open(status_file) as f:
|
|
return json.load(f)
|
|
|
|
def load_events() -> List[Dict]:
|
|
events_file = Path("/opt/biggfish/src/data/krystie-events.json")
|
|
if not events_file.exists():
|
|
return []
|
|
with open(events_file) as f:
|
|
data = json.load(f)
|
|
return data.get('events', [])
|
|
|
|
def format_currency(val: float) -> str:
|
|
"""Format with K/M suffix for readability"""
|
|
if abs(val) >= 1_000_000:
|
|
return f"${val/1_000_000:.2f}M"
|
|
elif abs(val) >= 1_000:
|
|
return f"${val/1_000:.1f}K"
|
|
else:
|
|
return f"${val:.2f}"
|
|
|
|
def format_pnl(val: float, pct: Optional[float] = None) -> str:
|
|
"""Format P&L with color emoji"""
|
|
emoji = "🟢" if val >= 0 else "🔴"
|
|
base = f"{emoji} {format_currency(val)}"
|
|
if pct is not None:
|
|
base += f" ({pct:+.2f}%)"
|
|
return base
|
|
|
|
def get_position_summary(positions: List[Dict]) -> str:
|
|
"""Summarize positions in a compact way"""
|
|
if not positions:
|
|
return "No positions"
|
|
|
|
total_unrealized = sum(p.get('unrealized_pnl', 0) for p in positions)
|
|
winning = [p for p in positions if p.get('unrealized_pnl', 0) > 0]
|
|
losing = [p for p in positions if p.get('unrealized_pnl', 0) < 0]
|
|
|
|
parts = []
|
|
if winning:
|
|
parts.append(f"{len(winning)} ✅")
|
|
if losing:
|
|
parts.append(f"{len(losing)} ❌")
|
|
|
|
status = " | ".join(parts) if parts else f"{len(positions)} flat"
|
|
return f"{status} → {format_currency(total_unrealized)}"
|
|
|
|
def get_top_movers(positions: List[Dict], limit: int = 3) -> List[str]:
|
|
"""Get top winning and losing positions"""
|
|
if not positions:
|
|
return []
|
|
|
|
sorted_pos = sorted(positions, key=lambda p: p.get('unrealized_pnl', 0), reverse=True)
|
|
lines = []
|
|
|
|
# Top winners
|
|
for p in sorted_pos[:limit]:
|
|
pnl = p.get('unrealized_pnl', 0)
|
|
if pnl > 0:
|
|
lines.append(f" ✅ {p['symbol']}: {format_currency(pnl)}")
|
|
|
|
# Top losers
|
|
for p in sorted_pos[-limit:]:
|
|
pnl = p.get('unrealized_pnl', 0)
|
|
if pnl < 0:
|
|
lines.append(f" ❌ {p['symbol']}: {format_currency(pnl)}")
|
|
|
|
return lines
|
|
|
|
def get_recent_trades(events: List[Dict], hours: int = 24) -> List[Dict]:
|
|
"""Get trades from last N hours"""
|
|
cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)
|
|
trades = []
|
|
|
|
for event in events:
|
|
if event.get('type') != 'trade':
|
|
continue
|
|
|
|
ts_str = event.get('timestamp')
|
|
if not ts_str:
|
|
continue
|
|
|
|
try:
|
|
ts = datetime.fromisoformat(ts_str.replace('Z', '+00:00'))
|
|
if ts >= cutoff:
|
|
trades.append(event)
|
|
except:
|
|
continue
|
|
|
|
return trades
|
|
|
|
def format_learning_insight(learning: Dict) -> str:
|
|
"""Translate learning metrics into plain English"""
|
|
epsilon = learning.get('rl_epsilon', 1.0)
|
|
generation = learning.get('ga_generation', 0)
|
|
fitness = learning.get('ga_best_fitness', 0)
|
|
|
|
# Epsilon interpretation
|
|
if epsilon > 0.5:
|
|
mode = "🎲 Exploring heavily"
|
|
elif epsilon > 0.2:
|
|
mode = "🔀 Balanced exploration"
|
|
elif epsilon > 0.05:
|
|
mode = "🎯 Mostly exploiting"
|
|
else:
|
|
mode = "🔒 Pure exploitation"
|
|
|
|
# Fitness interpretation
|
|
if fitness > 20:
|
|
perf = "Excellent"
|
|
elif fitness > 10:
|
|
perf = "Good"
|
|
elif fitness > 5:
|
|
perf = "Developing"
|
|
else:
|
|
perf = "Early stage"
|
|
|
|
return f"{mode} | Gen {generation} ({perf})"
|
|
|
|
def format_daily_report(status: Dict, events: List[Dict]) -> str:
|
|
"""Format a clean, scannable daily report"""
|
|
portfolio = status.get('portfolio', {})
|
|
positions = status.get('positions', [])
|
|
learning = status.get('learning', {})
|
|
config = status.get('config', {})
|
|
markets = status.get('markets', {})
|
|
|
|
equity = portfolio.get('equity', 0)
|
|
initial = config.get('initial_capital', 200000)
|
|
target = config.get('target_capital', 250000)
|
|
day_pnl = portfolio.get('day_pnl', 0)
|
|
day_pnl_pct = portfolio.get('day_pnl_pct', 0)
|
|
total_pnl = equity - initial
|
|
total_pnl_pct = (total_pnl / initial * 100) if initial > 0 else 0
|
|
|
|
# Recent trades
|
|
recent_trades = get_recent_trades(events, hours=24)
|
|
wins = [t for t in recent_trades if t.get('pnl', 0) > 0]
|
|
losses = [t for t in recent_trades if t.get('pnl', 0) < 0]
|
|
trade_pnl = sum(t.get('pnl', 0) for t in recent_trades)
|
|
|
|
lines = []
|
|
lines.append("🐟 <b>BIGGFISH Daily Report</b>")
|
|
lines.append(f"📅 {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}")
|
|
lines.append("")
|
|
|
|
# === PORTFOLIO SNAPSHOT ===
|
|
lines.append(f"💰 <b>{format_currency(equity)}</b> equity")
|
|
lines.append(f"📊 Today: {format_pnl(day_pnl, day_pnl_pct)}")
|
|
lines.append(f"📈 Total: {format_pnl(total_pnl, total_pnl_pct)}")
|
|
|
|
# Progress to goal
|
|
progress = (equity - initial) / (target - initial) * 100 if target > initial else 0
|
|
if progress < 0:
|
|
progress_text = f"⚠️ <b>Down {abs(progress):.1f}%</b> from start"
|
|
elif progress >= 100:
|
|
progress_text = f"🎯 <b>GOAL REACHED!</b>"
|
|
else:
|
|
progress_text = f"🎯 {progress:.1f}% to goal ({format_currency(target - equity)} left)"
|
|
lines.append(progress_text)
|
|
lines.append("")
|
|
|
|
# === TODAY'S ACTIVITY ===
|
|
lines.append("<b>📋 Today's Trading</b>")
|
|
if recent_trades:
|
|
win_rate = (len(wins) / len(recent_trades) * 100) if recent_trades else 0
|
|
lines.append(f" {len(recent_trades)} trades | {win_rate:.0f}% win rate")
|
|
lines.append(f" {format_pnl(trade_pnl)}")
|
|
|
|
# Show significant trades
|
|
significant = sorted(recent_trades, key=lambda t: abs(t.get('pnl', 0)), reverse=True)[:3]
|
|
for t in significant:
|
|
pnl = t.get('pnl', 0)
|
|
if abs(pnl) > 5: # Only show trades > $5 P&L
|
|
emoji = "✅" if pnl > 0 else "❌"
|
|
symbol = t.get('symbol', '?')
|
|
lines.append(f" {emoji} {symbol}: {format_currency(pnl)}")
|
|
else:
|
|
lines.append(" No trades today")
|
|
lines.append("")
|
|
|
|
# === OPEN POSITIONS ===
|
|
lines.append(f"<b>📊 Positions: {len(positions)}</b>")
|
|
if positions:
|
|
lines.append(f" {get_position_summary(positions)}")
|
|
|
|
# Show top movers
|
|
movers = get_top_movers(positions, limit=2)
|
|
if movers:
|
|
lines.extend(movers)
|
|
else:
|
|
lines.append(" All flat")
|
|
lines.append("")
|
|
|
|
# === LEARNING STATUS ===
|
|
lines.append("<b>🧠 Learning</b>")
|
|
lines.append(f" {format_learning_insight(learning)}")
|
|
lines.append("")
|
|
|
|
# === MARKET STATUS ===
|
|
stock_status = "🟢 Open" if markets.get('stocks') == 'OPEN' else "🔴 Closed"
|
|
forex_status = "🟢 Open" if markets.get('forex') == 'OPEN' else "🔴 Closed"
|
|
lines.append(f"<b>🏦 Markets:</b> Stocks {stock_status} | Forex {forex_status}")
|
|
|
|
return "\n".join(lines)
|
|
|
|
def main():
|
|
status = load_status()
|
|
events = load_events()
|
|
|
|
if not status:
|
|
print("❌ Could not load status data")
|
|
sys.exit(1)
|
|
|
|
report = format_daily_report(status, events)
|
|
print(report)
|
|
|
|
if __name__ == '__main__':
|
|
main()
|