#!/usr/bin/env python3 """展示页面后端 - 提供实时价格+AI分析数据API""" from http.server import HTTPServer, SimpleHTTPRequestHandler import json, urllib.request, os _json = json # 防止UnboundLocalError SHOWCASE_DATA = os.path.join(os.path.dirname(os.path.abspath(__file__)), "showcase_data.json") import time as _time _cache = {} # key -> (data, timestamp) def cached_fetch(key, url, ttl=1800, headers=None): """带缓存的HTTP请求，ttl秒内返回缓存""" if key in _cache and _time.time() - _cache[key][1] < ttl: return _cache[key][0] hdrs = {'User-Agent': 'Mozilla/5.0'} if headers: hdrs.update(headers) req = urllib.request.Request(url, headers=hdrs) r = urllib.request.urlopen(req, timeout=10) data = _json.loads(r.read()) _cache[key] = (data, _time.time()) return data class Handler(SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=os.path.dirname(os.path.abspath(__file__)), **kwargs) def do_GET(self): if self.path == '/api/prices': self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Cache-Control', 'no-store') self.end_headers() prices = {} for c in ['BTC_USDT', 'ETH_USDT', 'SOL_USDT']: try: r = urllib.request.urlopen(f"https://api.gateio.ws/api/v4/futures/usdt/tickers?contract={c}", timeout=5) d = _json.loads(r.read())[0] prices[c.split('_')[0]] = { 'price': float(d['last']), 'change': float(d['change_percentage']), 'high': float(d['high_24h']), 'low': float(d['low_24h']), 'volume': float(d.get('volume_24h', 0)) } except: pass self.wfile.write(_json.dumps(prices).encode()) elif self.path == '/api/calendar': self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Cache-Control', 'max-age=1800') self.end_headers() try: from datetime import datetime, timezone, timedelta now = datetime.now(timezone.utc) from_d = (now - timedelta(hours=24)).strftime('%Y-%m-%dT00:00:00.000Z') to_d = (now + timedelta(days=5)).strftime('%Y-%m-%dT00:00:00.000Z') url = f"https://economic-calendar.tradingview.com/events?from={from_d}&to={to_d}&countries=US,JP,EU,GB" cal = cached_fetch('calendar_tv', url, ttl=1800, headers={ 'Origin': 'https://www.tradingview.com' }) events = [] country_map = {'US':'USD','JP':'JPY','EU':'EUR','GB':'GBP'} flag_map = {'US':'🇺🇸','JP':'🇯🇵','EU':'🇪🇺','GB':'🇬🇧'} cn_map = { 'Fed Interest Rate Decision': '美联储利率决议', 'Fed Press Conference': '鲍威尔新闻发布会', 'FOMC Minutes': 'FOMC会议纪要', 'BoJ Interest Rate Decision': '日本央行利率决议', 'BoJ Press Conference': '日本央行发布会', 'BoE Interest Rate Decision': '英国央行利率决议', 'BoE Press Conference': '英国央行发布会', 'ECB Interest Rate Decision': '欧央行利率决议', 'ECB Press Conference': '欧央行发布会', 'Deposit Facility Rate': '欧央行存款利率', 'GDP Growth Rate QoQ Adv': '美国GDP季率(初值)', 'GDP Growth Rate QoQ Flash': '欧元区GDP季率', 'GDP Growth Rate YoY Flash': '欧元区GDP年率', 'GDP Growth Rate QoQ 2nd Est': '美国GDP季率(修正)', 'Core PCE Price Index MoM': '核心PCE物价月率', 'Core PCE Price Index YoY': '核心PCE物价年率', 'PCE Price Index MoM': 'PCE物价月率', 'PCE Price Index YoY': 'PCE物价年率', 'Personal Spending MoM': '个人支出月率', 'Personal Income MoM': '个人收入月率', 'Durable Goods Orders MoM': '耐用品订单月率', 'Building Permits Prel': '营建许可(初值)', 'Building Permits': '营建许可', 'Housing Starts': '新屋开工', 'Existing Home Sales': '成屋销售', 'New Home Sales': '新屋销售', 'Inflation Rate YoY Flash': '欧元区通胀年率', 'Inflation Rate MoM': '通胀月率', 'Inflation Rate YoY': '通胀年率', 'Core Inflation Rate YoY Flash': '欧元区核心通胀年率', 'Consumer Confidence': '消费者信心指数', 'Michigan Consumer Sentiment Final': '密歇根消费者信心(终值)', 'Michigan Consumer Sentiment Prel': '密歇根消费者信心(初值)', 'CB Consumer Confidence': 'CB消费者信心指数', 'Nonfarm Payrolls': '非农就业人数', 'Unemployment Rate': '失业率', 'ADP Employment Change': 'ADP就业人数', 'Initial Jobless Claims': '初请失业金人数', 'Continuing Jobless Claims': '续请失业金人数', 'JOLTs Job Openings': 'JOLTS职位空缺', 'CPI YoY': 'CPI年率', 'CPI MoM': 'CPI月率', 'Core CPI MoM': '核心CPI月率', 'Core CPI YoY': '核心CPI年率', 'PPI MoM': 'PPI月率', 'PPI YoY': 'PPI年率', 'Retail Sales MoM': '零售销售月率', 'Core Retail Sales MoM': '核心零售销售月率', 'ISM Manufacturing PMI': 'ISM制造业PMI', 'ISM Non-Manufacturing PMI': 'ISM非制造业PMI', 'ISM Manufacturing Prices': 'ISM制造业价格', 'S&P Global Manufacturing PMI Final': '标普制造业PMI(终值)', 'S&P Global Services PMI Final': '标普服务业PMI(终值)', 'Industrial Production MoM': '工业产出月率', 'Trade Balance': '贸易帐', 'Current Account': '经常帐', 'Import Prices MoM': '进口物价月率', 'Export Prices MoM': '出口物价月率', 'Treasury Currency Report': '财政部货币报告', 'President Trump Speaks': '特朗普讲话', 'Monetary Policy Statement': '货币政策声明', } def to_cn(title, country): flag = flag_map.get(country, '🌍') if title in cn_map: return flag + ' ' + cn_map[title] return flag + ' ' + title cal_list = cal.get('result', cal) if isinstance(cal, dict) else cal for e in (cal_list if isinstance(cal_list, list) else []): imp = e.get('importance', 0) if imp < 1: continue # 只要高影响力 try: raw_date = e.get('date') or e.get('time') or '' t = datetime.fromisoformat(raw_date.replace('Z','+00:00')) if raw_date else None if not t: continue hours = (t - now).total_seconds() / 3600 if hours > -24 and hours < 120: actual = e.get('actual') forecast = e.get('forecast') previous = e.get('previous') events.append({ 'title': to_cn(e.get('title',''), e.get('country','')), 'title_en': e.get('title',''), 'country': country_map.get(e.get('country',''), e.get('country','')), 'date': raw_date, 'hours_away': round(hours, 1), 'impact': 'High', 'actual': str(actual) if actual is not None else '', 'forecast': str(forecast) if forecast is not None else '', 'previous': str(previous) if previous is not None else '' }) except: pass events.sort(key=lambda x: x['hours_away']) self.wfile.write(_json.dumps(events[:20]).encode()) except Exception as ex: self.wfile.write(_json.dumps({"error": str(ex)}).encode()) elif self.path == '/api/stocks': self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Cache-Control', 'max-age=300') self.end_headers() stocks = {} for sym, name in [('%5EGSPC', 'S&P500'), ('%5EIXIC', 'NASDAQ'), ('%5EDJI', 'DOW')]: try: d = cached_fetch(f'stock_{sym}', f"https://query1.finance.yahoo.com/v8/finance/chart/{sym}?interval=1d&range=2d", ttl=300) res = d['chart']['result'][0] closes = res['indicators']['quote'][0]['close'] curr = closes[-1] prev = closes[-2] if len(closes) > 1 else curr if curr and prev: stocks[name] = { 'price': round(curr, 2), 'change': round((curr - prev) / prev * 100, 2), 'status': 'open' if res['meta'].get('regularMarketTime', 0) > 0 else 'closed' } except: pass self.wfile.write(_json.dumps(stocks).encode()) elif self.path == '/api/ai': self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Cache-Control', 'no-store') self.end_headers() try: import json, base64, os as _os with open(SHOWCASE_DATA) as f: data = _json.loads(f.read()) # K线截图改用URL加载（不嵌base64，大王5/12：733KB太大手机加载慢） if data.get('chartPaths'): for cp in data['chartPaths']: fpath = cp.get('path','') fname = _os.path.basename(fpath) cp['url'] = f'/api/chart/{fname}' cp.pop('base64', None) # 删除旧的base64 self.wfile.write(_json.dumps(data).encode()) except: import traceback; traceback.print_exc() self.wfile.write(b'{"error":"no data yet"}') elif self.path == '/api/sniper': self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Cache-Control', 'no-store') self.end_headers() import json result = {} try: with open('/root/.openclaw/workspace/strategies/sniper_state.json') as f: state = _json.loads(f.read()) pos = state.get('positions', {}) # 前端需要array不是dict # 精简positions：aiHistory只保留最近3条（大王5/12：原始数据太大手机加载不了） poslist = list(pos.values()) if isinstance(pos, dict) else pos for p in poslist: if 'aiHistory' in p: p['aiHistory'] = p['aiHistory'][-3:] result['positions'] = poslist result['stats'] = state.get('stats', {}) # watchPool只发摘要（大王5/12：完整338KB太大手机加载不了） wp = state.get('watchPool', {}) result['watchPool'] = {k: {'watchPrice': v.get('watchPrice'), 'reason': (v.get('reason','')[:60]), 'addedAt': v.get('addedAt')} for k,v in (list(wp.items())[:20])} except: pass try: # 大王5/16：改读sniper_trades.jsonl（真实成交pnl），不再读reviews（复盘已关+旧数据pnl=0） trades_file = '/root/.openclaw/workspace/strategies/sniper_trades.jsonl' import os if os.path.exists(trades_file): with open(trades_file) as f: lines = f.readlines() # 只取CLOSE类型的最近20条 closes = [] for line in reversed(lines): try: d = _json.loads(line.strip()) if d.get('type') == 'CLOSE': closes.append({ 'time': d.get('time',''), 'sym': d.get('sym',''), 'direction': d.get('direction','LONG'), 'entry': d.get('entry',0), 'exit': d.get('exit',0), 'pnl': str(round(float(d.get('pnl',0)), 2)), 'pnlPct': str(d.get('pnlPct',0)), 'reason': d.get('reason',''), 'holdHours': d.get('holdHours','0'), }) if len(closes) >= 20: break except: continue result['reviews'] = list(reversed(closes)) # 时间正序 else: result['reviews'] = [] except: result['reviews'] = [] try: result['learned'] = [] # 不再发learned（大王5/12：没用还占空间） except: pass try: with open('/root/.openclaw/workspace/strategies/sniper_ai_log.json') as f: logs = _json.loads(f.read()) result['aiLogs'] = logs[:10] # 只发最近10条（大王5/12：50条太多） except: result['aiLogs'] = [] # 币安余额（通过Node调用签名API） try: import subprocess bal_result = subprocess.run(['node', '--input-type=module', '-e', """ import { getBalance } from '/root/.openclaw/workspace/strategies/exchange_binance.mjs'; const b = await getBalance(); console.log(JSON.stringify(b)); """], capture_output=True, text=True, timeout=10) if bal_result.stdout.strip(): result['balance'] = _json.loads(bal_result.stdout.strip()) except: result['balance'] = {} # 币安持仓浮盈（通过Node） try: pos_result = subprocess.run(['node', '--input-type=module', '-e', """ import { getPositions } from '/root/.openclaw/workspace/strategies/exchange_binance.mjs'; const positions = await getPositions(); const active = positions.filter(p => parseFloat(p.size) !== 0); let totalPnl = 0; const list = active.map(p => { const pnl = parseFloat(p.pnl || 0); totalPnl += pnl; return { sym: p.symbol, pnl: pnl.toFixed(3), side: p.side || 'LONG', entry: p.entryPrice, mark: p.markPrice }; }); console.log(JSON.stringify({ totalPnl, positions: list })); """], capture_output=True, text=True, timeout=10) if pos_result.stdout.strip(): result['livePnl'] = _json.loads(pos_result.stdout.strip()) except: result['livePnl'] = {} # 进程状态 import subprocess procs = {} for name, pattern in [('sniper','altcoin_sniper'),('signal_d','signal_d_scanner'),('watchdog','watchdog.sh'),('bot','sniper_bot')]: try: r = subprocess.run(['pgrep','-f',pattern], capture_output=True, text=True, timeout=2) procs[name] = {'running': r.returncode==0, 'pid': r.stdout.strip().split('\n')[0] if r.returncode==0 else ''} except: procs[name] = {'running': False} result['processes'] = procs # 最近日志 try: import subprocess log_result = subprocess.run(['tail', '-30', '/tmp/altcoin_sniper.log'], capture_output=True, text=True, timeout=3) result['recentLog'] = log_result.stdout.strip().split('\n')[-30:] except: result['recentLog'] = [] # 挂单（从state里读pending） try: pending = {k:v for k,v in (state.get('positions',{})).items() if v.get('pendingOrder')} result['pendingOrders'] = pending except: result['pendingOrders'] = {} self.wfile.write(_json.dumps(result).encode()) elif self.path == '/sniper' or self.path == '/sniper.html': self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.send_header('Cache-Control', 'no-store') self.end_headers() dash = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'sniper_dashboard.html') with open(dash, 'rb') as f: self.wfile.write(f.read()) elif self.path.startswith('/api/chart/'): # 返回K线截图 /api/chart/BTCUSDT_4h.png fname = self.path.split('/api/chart/')[-1] # 安全检查：只允许字母数字下划线点 import re if not re.match(r'^[A-Z0-9a-z_]+\.png$', fname): self.send_error(400, 'Invalid filename') return fpath = f'/tmp/charts/{fname}' if os.path.exists(fpath): self.send_response(200) self.send_header('Content-Type', 'image/png') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Cache-Control', 'max-age=60') self.end_headers() with open(fpath, 'rb') as f: self.wfile.write(f.read()) else: self.send_error(404, 'Chart not found') elif self.path == '/' or self.path == '/index.html': # 默认返回showcase.html import os as _os2 showcase_html = _os2.path.join(_os2.path.dirname(_os2.path.abspath(__file__)), 'showcase.html') if _os2.path.exists(showcase_html): self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Cache-Control', 'no-store') self.end_headers() with open(showcase_html, 'rb') as f: self.wfile.write(f.read()) else: self.send_error(404, 'showcase.html not found') else: super().do_GET() print("Showcase server on :8091") import socketserver class ReusableHTTPServer(HTTPServer): allow_reuse_address = True allow_reuse_port = True ReusableHTTPServer(('0.0.0.0', 8091), Handler).serve_forever()