#!/usr/bin/env python3 """ XTZH AI定价引擎服务 NAC公链 - XTZH资产稳定币当日铸造价格计算 基于NAC宪法权重模型：货币层40% + 黄金层10% + 商品层50% 黄金信任锚覆盖率：125% """ import asyncio import json import logging import time import math from datetime import datetime, timezone from typing import Optional import urllib.request import urllib.error from http.server import HTTPServer, BaseHTTPRequestHandler import threading # ============================================================================ # 日志配置 # ============================================================================ logging.basicConfig( level=logging.INFO, format='%(asctime)s [XTZH-AI] %(levelname)s %(message)s', handlers=[ logging.FileHandler('/var/log/xtzh-pricing.log'), logging.StreamHandler() ] ) logger = logging.getLogger('xtzh-pricing') # ============================================================================ # NAC宪法权重常量（来自constants.rs） # ============================================================================ W_FX_BASE = 4000 # 货币层权重基准 40.00% W_AU_BASE = 1000 # 黄金层权重基准 10.00% W_COM_BASE = 5000 # 商品层权重基准 50.00% W_AU_MIN = 500 # 黄金层权重最小值 5.00% W_AU_MAX = 2000 # 黄金层权重最大值 20.00% WEIGHT_SUM = 10000 # 权重和 100.00% GOLD_TRUST_ANCHOR_COVERAGE = 12500 # 黄金信任锚覆盖率 125.00% # SDR货币篮子权重（IMF 2022-2026） SDR_BASKET = { 'USD': 0.4332, # 43.32% 'EUR': 0.2923, # 29.23% 'CNY': 0.1059, # 10.59% 'JPY': 0.0784, # 7.84% 'GBP': 0.0802, # 8.02% # 其余0.1%忽略 } # ============================================================================ # 数据缓存 # ============================================================================ _cache = { 'data': None, 'timestamp': 0, 'ttl': 300 # 5分钟缓存 } # ============================================================================ # 数据获取函数 # ============================================================================ def fetch_url(url, timeout=10): """获取URL内容""" try: req = urllib.request.Request(url, headers={ 'User-Agent': 'Mozilla/5.0 (NAC-XTZH-Pricing/1.0)', 'Accept': 'application/json' }) with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode('utf-8')) except Exception as e: logger.warning(f"获取 {url} 失败: {e}") return None def fetch_yahoo_price(symbol): """从Yahoo Finance获取价格""" import urllib.parse encoded = urllib.parse.quote(symbol) url = f"https://query1.finance.yahoo.com/v8/finance/chart/{encoded}?interval=1d&range=1d" data = fetch_url(url) if data and data.get('chart', {}).get('result'): return data['chart']['result'][0]['meta'].get('regularMarketPrice') return None def fetch_sdr_rates(): """获取SDR汇率（1 XDR = ? 各货币）""" data = fetch_url("https://open.er-api.com/v6/latest/XDR") if data and data.get('rates'): rates = data['rates'] return { 'USD': rates.get('USD', 1.374498), 'EUR': rates.get('EUR', 1.166327), 'GBP': rates.get('GBP', 1.017137), 'JPY': rates.get('JPY', 215.268845), 'CNY': rates.get('CNY', 9.444808), 'updated': data.get('time_last_update_utc', '') } # 备用：使用最近已知值 logger.warning("SDR汇率获取失败，使用备用数据") return { 'USD': 1.374498, 'EUR': 1.166327, 'GBP': 1.017137, 'JPY': 215.268845, 'CNY': 9.444808, 'updated': 'fallback' } def fetch_gold_price_usd(): """获取黄金价格（USD/盎司）""" # 主源：Yahoo Finance GC=F price = fetch_yahoo_price("GC=F") if price and price > 0: logger.info(f"黄金价格（Yahoo GC=F）: ${price}") return price, "Yahoo Finance GC=F" # 备源：Binance PAXG（黄金代币） data = fetch_url("https://api.binance.com/api/v3/ticker/price?symbol=PAXGUSDT") if data and data.get('price'): price = float(data['price']) logger.info(f"黄金价格（Binance PAXG）: ${price}") return price, "Binance PAXG" # 备源：CoinGecko data = fetch_url("https://api.coingecko.com/api/v3/simple/price?ids=pax-gold&vs_currencies=usd") if data and data.get('pax-gold', {}).get('usd'): price = float(data['pax-gold']['usd']) logger.info(f"黄金价格（CoinGecko PAXG）: ${price}") return price, "CoinGecko PAXG" # 最终备用 logger.warning("黄金价格获取失败，使用备用值") return 2650.0, "fallback" def fetch_market_indicators(): """获取市场指标（VIX、DXY、WTI等）""" indicators = {} # VIX波动率指数 vix = fetch_yahoo_price("^VIX") indicators['vix'] = vix if vix else 20.0 # DXY美元指数 dxy = fetch_yahoo_price("DX-Y.NYB") if not dxy: dxy = fetch_yahoo_price("DX=F") indicators['dxy'] = dxy if dxy else 104.0 # WTI原油 wti = fetch_yahoo_price("CL=F") indicators['wti'] = wti if wti else 75.0 # 铜期货 copper = fetch_yahoo_price("HG=F") indicators['copper'] = copper if copper else 4.2 logger.info(f"市场指标: VIX={indicators['vix']}, DXY={indicators['dxy']}, WTI={indicators['wti']}, 铜={indicators['copper']}") return indicators # ============================================================================ # XTZH AI定价引擎（核心计算） # ============================================================================ def calculate_fx_layer(sdr_rates): """ 计算货币层价值（FX Layer） 1 XTZH 对应的 SDR 价值的货币层部分 使用SDR货币篮子权重计算 """ # 1 SDR = 1 XTZH（基础锚定） # 货币层价值 = SDR在各货币中的加权价值 # 以USD计价：1 SDR = 1.374 USD sdr_in_usd = sdr_rates['USD'] # 货币篮子验证：各货币权重之和应接近1 # USD 43.32% + EUR 29.23% + CNY 10.59% + JPY 7.84% + GBP 8.02% = 99% fx_value_usd = sdr_in_usd # 货币层直接等于SDR的USD价值 logger.info(f"货币层: 1 SDR = ${sdr_in_usd:.4f} USD") return fx_value_usd def calculate_gold_layer(gold_price_usd, sdr_rates): """ 计算黄金层价值（Gold Layer） 黄金信任锚覆盖率125%，黄金层权重10%（可在5%-20%动态调整） 黄金层价值 = 黄金价格 / 标准黄金单位 * 黄金覆盖率 标准：1 SDR ≈ 0.0005 盎司黄金（历史锚定） """ sdr_in_usd = sdr_rates['USD'] # 黄金层价值：以黄金价格调整SDR价值 # 黄金覆盖率125%意味着每1 XTZH背后有1.25倍的黄金价值支撑 # 黄金调整因子 = 当前黄金价格 / 基准黄金价格（2020年基准 $1800/oz） gold_base_price = 1800.0 # 2020年基准价格 gold_adjustment = gold_price_usd / gold_base_price # 黄金层贡献值（以USD计） # 黄金层权重10%，覆盖率125% gold_coverage_ratio = GOLD_TRUST_ANCHOR_COVERAGE / 10000.0 # 1.25 gold_layer_value = sdr_in_usd * gold_adjustment * gold_coverage_ratio logger.info(f"黄金层: 价格=${gold_price_usd:.2f}, 调整因子={gold_adjustment:.4f}, 层价值=${gold_layer_value:.4f}") return gold_layer_value def calculate_commodity_layer(indicators, sdr_rates): """ 计算商品层价值（Commodity Layer） 商品层权重50%，包含原油、铜、农产品等大宗商品 商品层价值 = SDR价值 * 商品综合调整因子 """ sdr_in_usd = sdr_rates['USD'] # 商品调整因子计算 # WTI原油：基准$75，权重30% wti = indicators.get('wti', 75.0) wti_base = 75.0 wti_factor = wti / wti_base # 铜价：基准$4.2/磅，权重25% copper = indicators.get('copper', 4.2) copper_base = 4.2 copper_factor = copper / copper_base # DXY美元指数：基准104，反向关系（美元强则商品弱） dxy = indicators.get('dxy', 104.0) dxy_base = 104.0 dxy_factor = dxy_base / dxy # 反向 # VIX风险调整：基准20，高VIX降低商品层价值 vix = indicators.get('vix', 20.0) vix_base = 20.0 vix_factor = vix_base / max(vix, 10.0) # 防止除零 # 综合商品调整因子（加权平均） commodity_adjustment = ( wti_factor * 0.30 + copper_factor * 0.25 + dxy_factor * 0.30 + vix_factor * 0.15 ) commodity_layer_value = sdr_in_usd * commodity_adjustment logger.info(f"商品层: WTI因子={wti_factor:.4f}, 铜因子={copper_factor:.4f}, DXY因子={dxy_factor:.4f}, VIX因子={vix_factor:.4f}") logger.info(f"商品层综合调整={commodity_adjustment:.4f}, 层价值=${commodity_layer_value:.4f}") return commodity_layer_value def calculate_dynamic_weights(indicators, gold_price_usd): """ 计算动态权重（AI调整） 基于市场状态动态调整三层权重 约束：w_fx + w_au + w_com = 10000（基点） 黄金层：500 <= w_au <= 2000 """ vix = indicators.get('vix', 20.0) dxy = indicators.get('dxy', 104.0) # 风险情绪调整：高VIX时增加黄金权重 # VIX > 30: 高风险，黄金权重上调 # VIX < 15: 低风险，黄金权重下调 if vix > 30: w_au = min(W_AU_MAX, W_AU_BASE + int((vix - 30) * 30)) elif vix < 15: w_au = max(W_AU_MIN, W_AU_BASE - int((15 - vix) * 25)) else: w_au = W_AU_BASE # 美元强弱调整：DXY高时货币层权重下调 if dxy > 110: fx_adjustment = -200 elif dxy < 95: fx_adjustment = 200 else: fx_adjustment = int((104 - dxy) * 15) w_fx = W_FX_BASE + fx_adjustment w_fx = max(2000, min(6000, w_fx)) # 限制在20%-60% # 商品层补足剩余权重 w_com = WEIGHT_SUM - w_fx - w_au w_com = max(2000, min(7000, w_com)) # 限制在20%-70% # 重新归一化确保总和=10000 total = w_fx + w_au + w_com if total != WEIGHT_SUM: diff = WEIGHT_SUM - total w_com += diff logger.info(f"动态权重: FX={w_fx/100:.1f}%, AU={w_au/100:.1f}%, COM={w_com/100:.1f}%") return w_fx, w_au, w_com def calculate_xtzh_mint_price(): """ 计算XTZH当日铸造价格（核心函数） 价格公式： P_XTZH = (w_fx * V_fx + w_au * V_au + w_com * V_com) / WEIGHT_SUM 其中： - V_fx: 货币层价值（SDR USD等价） - V_au: 黄金层价值（黄金调整后） - V_com: 商品层价值（大宗商品调整后） - w_fx, w_au, w_com: 动态权重（基点，总和10000） """ logger.info("=== 开始计算XTZH当日铸造价格 ===") # 1. 获取实时数据 sdr_rates = fetch_sdr_rates() gold_price_usd, gold_source = fetch_gold_price_usd() indicators = fetch_market_indicators() # 2. 计算三层价值 v_fx = calculate_fx_layer(sdr_rates) v_au = calculate_gold_layer(gold_price_usd, sdr_rates) v_com = calculate_commodity_layer(indicators, sdr_rates) # 3. 计算动态权重 w_fx, w_au, w_com = calculate_dynamic_weights(indicators, gold_price_usd) # 4. 加权计算最终价格（USD） price_usd = (w_fx * v_fx + w_au * v_au + w_com * v_com) / WEIGHT_SUM # 5. 计算各货币价格 sdr_in_usd = sdr_rates['USD'] price_sdr = price_usd / sdr_in_usd # XTZH/SDR price_cny = price_usd / sdr_rates['USD'] * sdr_rates['CNY'] # 通过SDR换算 price_eur = price_usd / sdr_rates['USD'] * sdr_rates['EUR'] # 6. 计算铸造比例（1 USD = ? XTZH） mint_ratio = 1.0 / price_usd # 1 USD可以铸造多少XTZH # 7. 计算黄金储备支撑（每1 XTZH背后的黄金价值） gold_backing_per_xtzh = (w_au / WEIGHT_SUM) * v_au * (GOLD_TRUST_ANCHOR_COVERAGE / 10000.0) result = { "price_usd": round(price_usd, 6), "price_sdr": round(price_sdr, 6), "price_cny": round(price_cny, 6), "price_eur": round(price_eur, 6), "mint_ratio": round(mint_ratio, 6), "gold_backing_usd": round(gold_backing_per_xtzh, 6), "weights": { "fx_layer": round(w_fx / 100, 2), "gold_layer": round(w_au / 100, 2), "commodity_layer": round(w_com / 100, 2) }, "components": { "fx_value_usd": round(v_fx, 6), "gold_value_usd": round(v_au, 6), "commodity_value_usd": round(v_com, 6) }, "market_data": { "gold_price_usd": round(gold_price_usd, 2), "gold_source": gold_source, "sdr_rates": { "usd": round(sdr_rates['USD'], 6), "eur": round(sdr_rates['EUR'], 6), "gbp": round(sdr_rates['GBP'], 6), "jpy": round(sdr_rates['JPY'], 6), "cny": round(sdr_rates['CNY'], 6) }, "vix": round(indicators.get('vix', 0), 2), "dxy": round(indicators.get('dxy', 0), 3), "wti_oil": round(indicators.get('wti', 0), 2), "copper": round(indicators.get('copper', 0), 4) }, "model_info": { "version": "1.0.0", "algorithm": "NAC Constitutional Weighted Pricing", "gold_trust_anchor_coverage": "125%", "weight_constraint": "w_fx + w_au + w_com = 10000bp", "gold_weight_range": "5.00% - 20.00%" }, "timestamp": datetime.now(timezone.utc).isoformat(), "next_update": "每5分钟更新一次" } logger.info(f"=== XTZH铸造价格计算完成 ===") logger.info(f" 价格: ${price_usd:.6f} USD / {price_sdr:.6f} SDR") logger.info(f" 权重: FX={w_fx/100:.1f}% AU={w_au/100:.1f}% COM={w_com/100:.1f}%") logger.info(f" 黄金支撑: ${gold_backing_per_xtzh:.6f} USD/XTZH") return result # ============================================================================ # HTTP服务器 # ============================================================================ def get_cached_price(): """获取缓存的价格数据""" now = time.time() if _cache['data'] is None or (now - _cache['timestamp']) > _cache['ttl']: logger.info("缓存过期，重新计算XTZH价格...") try: _cache['data'] = calculate_xtzh_mint_price() _cache['timestamp'] = now except Exception as e: logger.error(f"价格计算失败: {e}") if _cache['data'] is None: # 返回基础值 _cache['data'] = { "price_usd": 1.374498, "price_sdr": 1.0, "price_cny": 9.444808, "price_eur": 1.166327, "error": str(e), "timestamp": datetime.now(timezone.utc).isoformat() } return _cache['data'] class XTZHPricingHandler(BaseHTTPRequestHandler): def log_message(self, format, *args): logger.info(f"HTTP {args[0]} {args[1]} {args[2]}") def send_cors_headers(self): self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type, Authorization') self.send_header('Content-Type', 'application/json; charset=utf-8') def do_OPTIONS(self): self.send_response(200) self.send_cors_headers() self.end_headers() def do_GET(self): path = self.path.split('?')[0] if path == '/health': self.send_response(200) self.send_cors_headers() self.end_headers() resp = json.dumps({"status": "ok", "service": "xtzh-pricing", "version": "1.0.0"}) self.wfile.write(resp.encode()) elif path in ['/price', '/api/xtzh/price', '/api/v1/xtzh/price']: data = get_cached_price() self.send_response(200) self.send_cors_headers() self.end_headers() resp = json.dumps({"success": True, "data": data}, ensure_ascii=False) self.wfile.write(resp.encode('utf-8')) elif path in ['/price/simple', '/api/xtzh/price/simple']: data = get_cached_price() simple = { "price_usd": data.get("price_usd"), "price_sdr": data.get("price_sdr"), "price_cny": data.get("price_cny"), "price_eur": data.get("price_eur"), "gold_price_usd": data.get("market_data", {}).get("gold_price_usd"), "timestamp": data.get("timestamp") } self.send_response(200) self.send_cors_headers() self.end_headers() resp = json.dumps({"success": True, "data": simple}, ensure_ascii=False) self.wfile.write(resp.encode('utf-8')) elif path in ['/price/refresh', '/api/xtzh/price/refresh']: # 强制刷新缓存 _cache['timestamp'] = 0 data = get_cached_price() self.send_response(200) self.send_cors_headers() self.end_headers() resp = json.dumps({"success": True, "data": data, "refreshed": True}, ensure_ascii=False) self.wfile.write(resp.encode('utf-8')) else: self.send_response(404) self.send_cors_headers() self.end_headers() resp = json.dumps({"error": "Not found", "endpoints": ["/health", "/price", "/price/simple", "/price/refresh"]}) self.wfile.write(resp.encode()) def run_server(port=8709): """启动HTTP服务器""" server = HTTPServer(('0.0.0.0', port), XTZHPricingHandler) logger.info(f"XTZH AI定价服务启动，端口: {port}") logger.info(f"API端点: http://0.0.0.0:{port}/price") # 启动时预热缓存 logger.info("预热缓存，首次计算XTZH价格...") get_cached_price() server.serve_forever() if __name__ == '__main__': run_server(8709)