期货套保系统的行情展示是交易决策的重要支撑，但产业用户的行情需求与投机交易存在显著差异。产业用户更关注基差变动、期现价差、跨月价差等组合指标，而非单一合约的盘口深度。本文从产业场景出发，解析期货套保系统中行情展示优化的数据架构、计算引擎与前端渲染方案。
一、产业场景的行情数据模型
产业用户的行情需求需要重新设计数据模型，支撑多维度的组合行情计算：
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from decimal import Decimal
from datetime import datetime
from enum import Enum
class QuoteType(Enum):
FUTURES = "期货"
SPOT = "现货"
BASIS = "基差"
SPREAD = "价差"
@dataclass
class FuturesQuote:
"""期货行情"""
instrument: str
last_price: Decimal
bid_price: Decimal
ask_price: Decimal
bid_volume: int
ask_volume: int
volume: int
open_interest: int
timestamp: datetime
@property
def mid_price(self) –> Decimal:
return (self.bid_price + self.ask_price) / 2
@dataclass
class SpotQuote:
"""现货行情"""
commodity: str
region: str # 地区
grade: str # 品级
price: Decimal
source: str # 数据来源
timestamp: datetime
@dataclass
class BasisQuote:
"""基差行情"""
commodity: str
futures_instrument: str
spot_price: Decimal
futures_price: Decimal
basis: Decimal # 基差 = 现货 – 期货
basis_rate: Decimal # 基差率 = 基差 / 期货
timestamp: datetime
@dataclass
class SpreadQuote:
"""跨期/跨品种价差"""
spread_id: str
near_instrument: str
far_instrument: str
near_price: Decimal
far_price: Decimal
spread: Decimal # 价差 = 近月 – 远月
timestamp: datetime
class IndustryQuoteAggregator:
"""产业行情聚合器"""
def __init__(self):
self.futures_quotes: Dict[str, FuturesQuote] = {}
self.spot_quotes: Dict[str, SpotQuote] = {}
def update_futures(self, quote: FuturesQuote):
self.futures_quotes[quote.instrument] = quote
def update_spot(self, quote: SpotQuote):
key = f"{quote.commodity}_{quote.region}_{quote.grade}"
self.spot_quotes[key] = quote
def calculate_basis(
self,
commodity: str,
futures_instrument: str,
spot_region: str,
spot_grade: str
) –> Optional[BasisQuote]:
"""计算基差"""
futures = self.futures_quotes.get(futures_instrument)
spot_key = f"{commodity}_{spot_region}_{spot_grade}"
spot = self.spot_quotes.get(spot_key)
if not futures or not spot:
return None
basis = spot.price – futures.last_price
basis_rate = basis / futures.last_price if futures.last_price else Decimal(0)
return BasisQuote(
commodity=commodity,
futures_instrument=futures_instrument,
spot_price=spot.price,
futures_price=futures.last_price,
basis=basis,
basis_rate=basis_rate,
timestamp=datetime.now()
)
def calculate_spread(
self,
near_instrument: str,
far_instrument: str
) –> Optional[SpreadQuote]:
"""计算跨期价差"""
near = self.futures_quotes.get(near_instrument)
far = self.futures_quotes.get(far_instrument)
if not near or not far:
return None
return SpreadQuote(
spread_id=f"{near_instrument}–{far_instrument}",
near_instrument=near_instrument,
far_instrument=far_instrument,
near_price=near.last_price,
far_price=far.last_price,
spread=near.last_price – far.last_price,
timestamp=datetime.now()
)
# 使用示例
aggregator = IndustryQuoteAggregator()
# 模拟行情更新
aggregator.update_futures(FuturesQuote(
instrument="rb2603", last_price=Decimal("3850"),
bid_price=Decimal("3849"), ask_price=Decimal("3851"),
bid_volume=500, ask_volume=480, volume=125000, open_interest=850000,
timestamp=datetime.now()
))
aggregator.update_spot(SpotQuote(
commodity="螺纹钢", region="上海", grade="HRB400",
price=Decimal("3920"), source="我的钢铁网",
timestamp=datetime.now()
))
basis = aggregator.calculate_basis("螺纹钢", "rb2603", "上海", "HRB400")
print(f"基差: {basis.basis}, 基差率: {basis.basis_rate:.2%}")
产业行情模型支撑基差、价差等组合指标的实时计算。
二、行情计算引擎的性能优化
高频行情更新需要优化计算引擎，避免冗余计算：
from typing import Set, Callable, Dict, Any
from collections import defaultdict
from datetime import datetime
import time
class QuoteSubscription:
"""行情订阅管理"""
def __init__(self):
self.subscriptions: Dict[str, Set[Callable]] = defaultdict(set)
self.computed_quotes: Dict[str, Any] = {}
self.last_update: Dict[str, datetime] = {}
self.update_count = 0
def subscribe(self, quote_id: str, callback: Callable):
"""订阅行情"""
self.subscriptions[quote_id].add(callback)
def unsubscribe(self, quote_id: str, callback: Callable):
"""取消订阅"""
self.subscriptions[quote_id].discard(callback)
def notify(self, quote_id: str, data: Any):
"""通知订阅者"""
for callback in self.subscriptions.get(quote_id, []):
callback(quote_id, data)
class IncrementalQuoteEngine:
"""增量行情计算引擎"""
def __init__(self):
self.raw_quotes: Dict[str, Dict] = {}
self.computed_cache: Dict[str, Any] = {}
self.dependencies: Dict[str, Set[str]] = defaultdict(set)
self.subscription = QuoteSubscription()
def register_dependency(self, computed_id: str, source_ids: List[str]):
"""注册计算依赖关系"""
for source_id in source_ids:
self.dependencies[source_id].add(computed_id)
def update_raw_quote(self, quote_id: str, data: Dict):
"""更新原始行情"""
old_data = self.raw_quotes.get(quote_id)
# 检测是否有实质变化
if old_data and self._is_same(old_data, data):
return # 无变化，跳过
self.raw_quotes[quote_id] = data
# 触发依赖的计算更新
affected = self.dependencies.get(quote_id, set())
for computed_id in affected:
self._invalidate_cache(computed_id)
# 通知订阅者
self.subscription.notify(quote_id, data)
def _is_same(self, old: Dict, new: Dict) –> bool:
"""检查行情是否相同（忽略时间戳）"""
keys_to_compare = ['last_price', 'bid_price', 'ask_price']
return all(old.get(k) == new.get(k) for k in keys_to_compare)
def _invalidate_cache(self, computed_id: str):
"""使缓存失效"""
if computed_id in self.computed_cache:
del self.computed_cache[computed_id]
def get_computed_quote(
self,
computed_id: str,
compute_func: Callable
) –> Any:
"""获取计算行情（带缓存）"""
if computed_id not in self.computed_cache:
self.computed_cache[computed_id] = compute_func()
return self.computed_cache[computed_id]
# 性能测试
engine = IncrementalQuoteEngine()
# 注册基差计算依赖
engine.register_dependency("basis_rb", ["rb2603", "spot_rb_sh"])
# 模拟行情更新
start = time.time()
for i in range(10000):
engine.update_raw_quote("rb2603", {
"last_price": 3850 + i % 10,
"bid_price": 3849 + i % 10,
"ask_price": 3851 + i % 10,
"timestamp": datetime.now()
})
elapsed = time.time() – start
print(f"处理10000次行情更新耗时: {elapsed*1000:.2f}ms")
增量计算与缓存机制显著降低CPU开销。
三、前端行情展示的渲染优化
高频行情更新对前端渲染提出挑战，需采用虚拟化与节流策略：
from dataclasses import dataclass
from typing import List, Dict, Any, Optional
from datetime import datetime
import json
@dataclass
class QuoteDisplayConfig:
"""行情展示配置"""
max_rows: int = 50 # 最大显示行数
update_interval_ms: int = 100 # 更新间隔（毫秒）
highlight_duration_ms: int = 500 # 变化高亮持续时间
decimal_places: int = 2 # 小数位数
class QuoteDisplayManager:
"""行情展示管理器"""
def __init__(self, config: QuoteDisplayConfig):
self.config = config
self.display_buffer: List[Dict] = []
self.pending_updates: Dict[str, Dict] = {}
self.last_flush_time: Optional[datetime] = None
def queue_update(self, quote_id: str, data: Dict):
"""将更新加入队列"""
self.pending_updates[quote_id] = {
**data,
"quote_id": quote_id,
"update_time": datetime.now().isoformat()
}
def should_flush(self) –> bool:
"""判断是否应该刷新显示"""
if not self.last_flush_time:
return True
elapsed_ms = (datetime.now() – self.last_flush_time).total_seconds() * 1000
return elapsed_ms >= self.config.update_interval_ms
def flush_to_display(self) –> List[Dict]:
"""刷新到显示层"""
if not self.pending_updates:
return []
updates = list(self.pending_updates.values())
self.pending_updates.clear()
self.last_flush_time = datetime.now()
# 计算变化高亮
for update in updates:
update["highlights"] = self._calculate_highlights(update)
return updates
def _calculate_highlights(self, update: Dict) –> Dict[str, str]:
"""计算需要高亮的字段"""
highlights = {}
quote_id = update.get("quote_id")
# 查找历史数据比较
for item in self.display_buffer:
if item.get("quote_id") == quote_id:
if update.get("last_price", 0) > item.get("last_price", 0):
highlights["last_price"] = "up"
elif update.get("last_price", 0) < item.get("last_price", 0):
highlights["last_price"] = "down"
break
return highlights
def format_for_display(self, quote: Dict) –> Dict:
"""格式化用于显示"""
return {
"合约": quote.get("quote_id", ""),
"最新价": f"{quote.get('last_price', 0):,.{self.config.decimal_places}f}",
"买价": f"{quote.get('bid_price', 0):,.{self.config.decimal_places}f}",
"卖价": f"{quote.get('ask_price', 0):,.{self.config.decimal_places}f}",
"成交量": f"{quote.get('volume', 0):,}",
"持仓量": f"{quote.get('open_interest', 0):,}",
"更新时间": quote.get("update_time", ""),
"_highlights": quote.get("highlights", {})
}
# 模拟行情展示流程
config = QuoteDisplayConfig(update_interval_ms=100)
display_mgr = QuoteDisplayManager(config)
# 模拟快速行情更新
for i in range(50):
display_mgr.queue_update(f"rb260{i % 12 + 1}", {
"last_price": 3850 + i,
"bid_price": 3849 + i,
"ask_price": 3851 + i,
"volume": 125000 + i * 100,
"open_interest": 850000
})
# 刷新显示
updates = display_mgr.flush_to_display()
print(f"本次刷新 {len(updates)} 条行情")
# 格式化输出
for update in updates[:3]:
formatted = display_mgr.format_for_display(update)
print(formatted)
节流与批量更新策略保障前端渲染的流畅性。
四、自定义行情看板的配置架构
产业用户需要灵活配置个性化的行情看板：
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional
from enum import Enum
import json
class WidgetType(Enum):
QUOTE_TABLE = "行情表格"
BASIS_CHART = "基差图表"
SPREAD_MONITOR = "价差监控"
POSITION_OVERVIEW = "持仓概览"
@dataclass
class WidgetConfig:
"""组件配置"""
widget_id: str
widget_type: WidgetType
title: str
position: Dict[str, int] # {x, y, width, height}
data_source: List[str] # 数据源ID列表
settings: Dict[str, Any] = field(default_factory=dict)
@dataclass
class DashboardLayout:
"""看板布局"""
dashboard_id: str
name: str
widgets: List[WidgetConfig] = field(default_factory=list)
def add_widget(self, widget: WidgetConfig):
self.widgets.append(widget)
def remove_widget(self, widget_id: str):
self.widgets = [w for w in self.widgets if w.widget_id != widget_id]
def to_json(self) –> str:
return json.dumps({
"dashboard_id": self.dashboard_id,
"name": self.name,
"widgets": [
{
"widget_id": w.widget_id,
"widget_type": w.widget_type.value,
"title": w.title,
"position": w.position,
"data_source": w.data_source,
"settings": w.settings
}
for w in self.widgets
]
}, ensure_ascii=False, indent=2)
class DashboardManager:
"""看板管理器"""
def __init__(self):
self.dashboards: Dict[str, DashboardLayout] = {}
self.user_preferences: Dict[str, str] = {} # user_id -> dashboard_id
def create_dashboard(self, name: str) –> DashboardLayout:
import uuid
dashboard_id = str(uuid.uuid4())[:8]
dashboard = DashboardLayout(dashboard_id=dashboard_id, name=name)
self.dashboards[dashboard_id] = dashboard
return dashboard
def get_default_industry_dashboard(self) –> DashboardLayout:
"""获取产业用户默认看板"""
dashboard = self.create_dashboard("产业套保看板")
# 基差监控
dashboard.add_widget(WidgetConfig(
widget_id="w1",
widget_type=WidgetType.BASIS_CHART,
title="螺纹钢基差走势",
position={"x": 0, "y": 0, "width": 6, "height": 4},
data_source=["basis_rb_sh"],
settings={"period": "1M", "show_ma": True}
))
# 跨期价差
dashboard.add_widget(WidgetConfig(
widget_id="w2",
widget_type=WidgetType.SPREAD_MONITOR,
title="螺纹钢跨期价差",
position={"x": 6, "y": 0, "width": 6, "height": 4},
data_source=["rb2603", "rb2605", "rb2610"],
settings={"alert_threshold": 50}
))
# 行情表格
dashboard.add_widget(WidgetConfig(
widget_id="w3",
widget_type=WidgetType.QUOTE_TABLE,
title="关注合约行情",
position={"x": 0, "y": 4, "width": 12, "height": 6},
data_source=["rb2603", "rb2605", "hc2603", "i2605"],
settings={"columns": ["合约", "最新价", "涨跌", "基差", "持仓"]}
))
return dashboard
# 创建并配置看板
mgr = DashboardManager()
dashboard = mgr.get_default_industry_dashboard()
print("=== 产业套保看板配置 ===")
print(dashboard.to_json())
灵活的看板配置满足不同产业用户的个性化需求。
总结
期货套保系统的行情展示优化需建立面向产业场景的数据模型，支撑基差、价差等组合指标的实时计算；通过增量计算与缓存机制提升计算引擎性能；采用节流与批量更新策略保障前端渲染流畅性；并提供灵活的看板配置架构满足用户个性化需求。整套优化方案使行情展示更贴合产业用户的决策场景。
