Spaces:
Running
Running
Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 19)
da0ce7b verified | #!/usr/bin/env python3 | |
| """ | |
| Lightweight CLI to inspect EnhancementLayer modifiers + a quick fused probability proxy | |
| for a given symbol/exchange using public tickers (Binance/Kraken). | |
| Usage examples: | |
| python scripts/enhancement_cli.py --symbol BTCUSDT --exchange binance | |
| python scripts/enhancement_cli.py --symbol XBTUSD --exchange kraken | |
| Outputs: | |
| - Price, 24h change %, volume | |
| - Enhancement modifier + emotional state/phase/band/chakra | |
| - Simple fused probability + action bucket (proxy, not production router) | |
| This avoids needing full aureon_unified_ecosystem; uses public REST. | |
| """ | |
| from aureon_baton_link import link_system as _baton_link; _baton_link(__name__) | |
| import argparse | |
| import sys | |
| import math | |
| from typing import Optional, Tuple, List | |
| import requests | |
| # Ensure local imports work when run from repository root | |
| sys.path.append('.') | |
| try: | |
| from aureon_enhancements import EnhancementLayer, get_emotional_color | |
| except Exception as exc: # pragma: no cover - optional deps | |
| print(f"β οΈ EnhancementLayer not available: {exc}") | |
| EnhancementLayer = None # type: ignore | |
| get_emotional_color = lambda x: '' # type: ignore | |
| def clamp(val: float, lo: float, hi: float) -> float: | |
| return max(lo, min(hi, val)) | |
| def synth_waveform(freq: float, samples: int = 32, sample_rate: int = 8000) -> List[float]: | |
| """Generate a short sine waveform preview for readability (not for audio).""" | |
| if freq <= 0 or samples <= 0: | |
| return [] | |
| out: List[float] = [] | |
| for n in range(samples): | |
| t = n / sample_rate | |
| out.append(math.sin(2 * math.pi * freq * t)) | |
| return out | |
| def sparkline(values: List[float]) -> str: | |
| """Render a tiny sparkline for quick visual read.""" | |
| if not values: | |
| return "" | |
| blocks = "βββββ βββ" | |
| lo, hi = min(values), max(values) | |
| span = hi - lo if hi != lo else 1.0 | |
| scaled = [int((v - lo) / span * (len(blocks) - 1)) for v in values] | |
| return ''.join(blocks[i] for i in scaled) | |
| def fetch_binance(symbol: str) -> Tuple[Optional[float], Optional[float], Optional[float]]: | |
| url = 'https://api.binance.com/api/v3/ticker/24hr' | |
| try: | |
| resp = requests.get(url, params={'symbol': symbol}, timeout=5) | |
| data = resp.json() | |
| if 'lastPrice' not in data: | |
| return None, None, None | |
| price = float(data.get('lastPrice', 0)) | |
| change_pct = float(data.get('priceChangePercent', 0)) | |
| volume = float(data.get('quoteVolume', 0)) | |
| return price, change_pct, volume | |
| except Exception: | |
| return None, None, None | |
| def fetch_kraken(symbol: str) -> Tuple[Optional[float], Optional[float], Optional[float]]: | |
| # Kraken expects pairs like XBTUSD, ETHUSD, SOLUSD | |
| pair = symbol.upper() | |
| url = 'https://api.kraken.com/0/public/Ticker' | |
| try: | |
| resp = requests.get(url, params={'pair': pair}, timeout=5) | |
| data = resp.json() | |
| if data.get('error'): | |
| return None, None, None | |
| result = data.get('result') or {} | |
| if not result: | |
| return None, None, None | |
| first = next(iter(result.values())) | |
| last = float(first.get('c', [0])[0]) | |
| open_price = float(first.get('o', last)) or last | |
| change_pct = ((last - open_price) / open_price * 100) if open_price else 0.0 | |
| volume = float(first.get('v', [0, 0])[1]) | |
| return last, change_pct, volume | |
| except Exception: | |
| return None, None, None | |
| def action_bucket(p: float) -> str: | |
| if p >= 0.70: | |
| return 'STRONG_BUY' | |
| if p >= 0.60: | |
| return 'BUY' | |
| if p >= 0.55: | |
| return 'SLIGHT_BUY' | |
| if p >= 0.45: | |
| return 'HOLD' | |
| if p >= 0.40: | |
| return 'SLIGHT_SELL' | |
| if p >= 0.30: | |
| return 'SELL' | |
| return 'STRONG_SELL' | |
| def main(): | |
| parser = argparse.ArgumentParser(description="Enhancement modifier + fused-prob proxy") | |
| parser.add_argument('--symbol', default='BTCUSDT', help='Symbol (e.g., BTCUSDT or XBTUSD)') | |
| parser.add_argument('--exchange', default='binance', choices=['binance', 'kraken'], help='Exchange to query') | |
| parser.add_argument('--waveform', action='store_true', help='Show a short waveform preview at the emotional frequency') | |
| parser.add_argument('--samples', type=int, default=32, help='Number of samples for waveform preview (default 32)') | |
| args = parser.parse_args() | |
| exchange = args.exchange.lower() | |
| symbol = args.symbol.upper() | |
| if exchange == 'binance': | |
| price, change_pct, volume = fetch_binance(symbol) | |
| else: | |
| price, change_pct, volume = fetch_kraken(symbol) | |
| if price is None: | |
| print(f"π« Could not fetch ticker for {symbol} on {exchange}") | |
| sys.exit(1) | |
| change_pct = change_pct or 0.0 | |
| volume = volume or 0.0 | |
| # Proxy lambda and coherence from price action | |
| lambda_value = clamp(change_pct / 100.0, -1.0, 1.0) | |
| coherence = clamp(abs(change_pct) / 10.0, 0.0, 1.0) # heuristic: larger move β higher coherence | |
| volatility = abs(change_pct) / 100.0 | |
| enh_modifier = 1.0 | |
| enh_state = 'Neutral' | |
| enh_phase = 'LOVE' | |
| enh_band = None | |
| enh_chakra = None | |
| enh_freq = None | |
| color = '' | |
| if EnhancementLayer: | |
| try: | |
| layer = EnhancementLayer() | |
| result = layer.get_unified_modifier( | |
| lambda_value=lambda_value, | |
| coherence=coherence, | |
| price=price, | |
| volume=volume, | |
| volatility=volatility, | |
| exchange=exchange.upper(), | |
| ) | |
| enh_modifier = result.trading_modifier | |
| enh_state = result.emotional_state | |
| enh_phase = result.cycle_phase | |
| enh_band = result.emotion_band | |
| enh_chakra = result.chakra_alignment | |
| enh_freq = result.emotional_frequency | |
| color = get_emotional_color(result.emotional_state) | |
| except Exception as exc: # pragma: no cover | |
| print(f"β οΈ Enhancement calc failed: {exc}") | |
| # Simple fused probability proxy | |
| base_prob = 0.5 + (change_pct / 100.0) * 0.5 # map -100..+100% to 0..1 with center 0.5 | |
| base_prob = clamp(base_prob, 0.0, 1.0) | |
| fused_prob = clamp(base_prob * enh_modifier, 0.0, 1.0) | |
| action = action_bucket(fused_prob) | |
| print(f"\nπ {exchange.upper()} {symbol}") | |
| print(f" Price: {price:.6f} | 24h Ξ: {change_pct:+.2f}% | Vol: {volume:.2f}") | |
| print(f" Ξ»: {lambda_value:+.3f} | Coherence: {coherence:.3f} | Volatility: {volatility:.3f}") | |
| print(f"\n⨠Enhancement Layer") | |
| print(f" Modifier: {enh_modifier:.3f}x | State: {enh_state} | Phase: {enh_phase} {color}") | |
| if enh_band: | |
| print(f" Emotion band: {enh_band}") | |
| if enh_chakra: | |
| print(f" Chakra alignment: {enh_chakra}") | |
| if args.waveform and enh_freq: | |
| wf = synth_waveform(enh_freq, samples=args.samples) | |
| print(f" Waveform @ {enh_freq:.1f}Hz (first {len(wf)} samples):") | |
| print(f" {sparkline(wf)}") | |
| if wf: | |
| # Show numeric preview for clarity | |
| preview = ', '.join(f"{v:+.2f}" for v in wf[:8]) | |
| print(f" samples: {preview} ...") | |
| print(f"\nπ§ Fused Prob (proxy)") | |
| print(f" Base: {base_prob:.3f} | Fused: {fused_prob:.3f} | Action: {action}") | |
| print() | |
| if __name__ == '__main__': | |
| main() | |