在多用户、多交易员并行的期现业务场景中，数据冲突是常见问题。基差风险管理系统的冲突预警模块，通过实时检测、智能识别与及时预警，帮助用户及时发现并处理数据冲突，确保数据一致性。本文将详细解析冲突预警模块的设计原理、检测机制与处理流程。
一、冲突类型与检测机制
期现业务中的冲突主要包括：数据修改冲突（多人同时修改同一数据）、匹配冲突（同一成交被匹配到多个合同）、权限冲突（无权限操作）、逻辑冲突（数据逻辑不一致）。
在快期-匹配宝系统中，冲突检测采用实时检测与定期扫描相结合的方式，实时检测在操作发生时立即检测，定期扫描定时检查历史数据。
from dataclasses import dataclass
from typing import List, Dict
from datetime import datetime
from enum import Enum
class ConflictType(Enum):
"""冲突类型"""
CONCURRENT_MODIFY = "并发修改"
DUPLICATE_MATCH = "重复匹配"
PERMISSION_VIOLATION = "权限违规"
LOGIC_INCONSISTENCY = "逻辑不一致"
@dataclass
class Conflict:
"""冲突记录"""
conflict_id: str
conflict_type: ConflictType
resource_type: str
resource_id: str
conflict_details: Dict
detected_at: datetime
severity: str # 'low', 'medium', 'high'
status: str # 'pending', 'resolved', 'ignored'
class ConflictDetector:
"""冲突检测器"""
def __init__(self):
self.lock_manager = LockManager()
self.match_validator = MatchValidator()
def detect_concurrent_modify(self, resource_id: str, user_id: str,
operation: str) –> Conflict:
"""检测并发修改冲突"""
# 检查资源是否被锁定
lock_info = self.lock_manager.get_lock(resource_id)
if lock_info and lock_info['user_id'] != user_id:
return Conflict(
conflict_id=self._generate_conflict_id(),
conflict_type=ConflictType.CONCURRENT_MODIFY,
resource_type=self._get_resource_type(resource_id),
resource_id=resource_id,
conflict_details={
"current_user": user_id,
"locked_by": lock_info['user_id'],
"lock_time": lock_info['locked_at'],
"operation": operation
},
detected_at=datetime.now(),
severity="high",
status="pending"
)
return None
def detect_duplicate_match(self, trade_id: str) –> List[Conflict]:
"""检测重复匹配冲突"""
# 查询该成交的所有匹配关系
matches = self.match_validator.get_matches_by_trade(trade_id)
conflicts = []
# 检查匹配数量是否超过成交数量
trade = self.get_trade(trade_id)
total_matched = sum(m.quantity for m in matches)
if total_matched > trade.quantity:
conflicts.append(Conflict(
conflict_id=self._generate_conflict_id(),
conflict_type=ConflictType.DUPLICATE_MATCH,
resource_type="trade_match",
resource_id=trade_id,
conflict_details={
"trade_quantity": trade.quantity,
"matched_quantity": total_matched,
"overlap": total_matched – trade.quantity,
"matches": [{"contract_id": m.contract_id, "quantity": m.quantity}
for m in matches]
},
detected_at=datetime.now(),
severity="high",
status="pending"
))
# 检查是否有重复匹配到同一合同
contract_matches = {}
for match in matches:
if match.contract_id not in contract_matches:
contract_matches[match.contract_id] = []
contract_matches[match.contract_id].append(match)
for contract_id, match_list in contract_matches.items():
if len(match_list) > 1:
conflicts.append(Conflict(
conflict_id=self._generate_conflict_id(),
conflict_type=ConflictType.DUPLICATE_MATCH,
resource_type="contract_match",
resource_id=contract_id,
conflict_details={
"trade_id": trade_id,
"duplicate_matches": len(match_list),
"match_ids": [m.match_id for m in match_list]
},
detected_at=datetime.now(),
severity="medium",
status="pending"
))
return conflicts
def detect_logic_inconsistency(self, resource_id: str) –> List[Conflict]:
"""检测逻辑不一致冲突"""
conflicts = []
# 检查合同执行数量是否超过合同总量
contract = self.get_contract(resource_id)
if contract.linked_quantity > contract.total_quantity:
conflicts.append(Conflict(
conflict_id=self._generate_conflict_id(),
conflict_type=ConflictType.LOGIC_INCONSISTENCY,
resource_type="contract",
resource_id=resource_id,
conflict_details={
"total_quantity": contract.total_quantity,
"linked_quantity": contract.linked_quantity,
"excess": contract.linked_quantity – contract.total_quantity
},
detected_at=datetime.now(),
severity="high",
status="pending"
))
return conflicts
冲突检测支持多种检测规则，用户可根据业务需求配置检测规则。
二、冲突预警的通知机制
冲突检测到后需要及时通知相关人员，预警通知包括：实时通知（冲突检测到立即通知）、分级通知（根据严重程度采用不同通知方式）、批量通知（批量冲突汇总通知）。
class ConflictAlertManager:
"""冲突预警管理器"""
def __init__(self):
self.notification_service = NotificationService()
self.alert_rules = {}
def trigger_conflict_alert(self, conflict: Conflict):
"""触发冲突预警"""
# 确定通知对象
recipients = self._get_alert_recipients(conflict)
# 生成预警消息
message = self._generate_alert_message(conflict)
# 根据严重程度选择通知方式
channels = self._select_notification_channels(conflict.severity)
# 发送通知
self.notification_service.send(
recipients=recipients,
subject=f"【冲突预警】{conflict.conflict_type.value}",
message=message,
channels=channels,
priority='high' if conflict.severity == 'high' else 'normal'
)
# 创建待办事项
self._create_todo_item(conflict)
def _generate_alert_message(self, conflict: Conflict) –> str:
"""生成预警消息"""
if conflict.conflict_type == ConflictType.CONCURRENT_MODIFY:
return f"""
检测到并发修改冲突
资源类型：{conflict.resource_type}
资源ID：{conflict.resource_id}
当前操作用户：{conflict.conflict_details['current_user']}
资源已被用户 {conflict.conflict_details['locked_by']} 锁定
锁定时间：{conflict.conflict_details['lock_time']}
请等待资源解锁后再操作，或联系锁定用户协调处理。
"""
elif conflict.conflict_type == ConflictType.DUPLICATE_MATCH:
return f"""
检测到重复匹配冲突
成交ID：{conflict.resource_id}
匹配数量：{conflict.conflict_details.get('matched_quantity', 0)}
成交数量：{conflict.conflict_details.get('trade_quantity', 0)}
超出数量：{conflict.conflict_details.get('overlap', 0)}
请检查并修正匹配关系，确保匹配数量不超过成交数量。
"""
else:
return f"""
检测到{conflict.conflict_type.value}冲突
资源类型：{conflict.resource_type}
资源ID：{conflict.resource_id}
冲突详情：{conflict.conflict_details}
请及时处理。
"""
def _select_notification_channels(self, severity: str) –> List[str]:
"""选择通知渠道"""
if severity == 'high':
return ['sms', 'wechat', 'email']
elif severity == 'medium':
return ['wechat', 'email']
else:
return ['email']
预警通知支持多通道并行发送，确保关键冲突能够及时触达。
三、冲突处理的流程
冲突检测到后需要建立处理流程，包括：冲突确认（确认冲突真实性）、冲突分析（分析冲突原因与影响）、冲突解决（采取解决措施）、冲突验证（验证解决效果）。
class ConflictResolver:
"""冲突解决器"""
def resolve_conflict(self, conflict_id: str, resolution: Dict) –> Dict:
"""解决冲突"""
conflict = self.get_conflict(conflict_id)
if conflict.status != 'pending':
raise ValueError(f"冲突 {conflict_id} 状态为 {conflict.status}，无法处理")
# 根据冲突类型选择解决方案
if conflict.conflict_type == ConflictType.CONCURRENT_MODIFY:
result = self._resolve_concurrent_modify(conflict, resolution)
elif conflict.conflict_type == ConflictType.DUPLICATE_MATCH:
result = self._resolve_duplicate_match(conflict, resolution)
elif conflict.conflict_type == ConflictType.LOGIC_INCONSISTENCY:
result = self._resolve_logic_inconsistency(conflict, resolution)
else:
result = {"success": False, "reason": "未知冲突类型"}
# 更新冲突状态
if result['success']:
self.update_conflict_status(conflict_id, 'resolved', resolution)
else:
self.update_conflict_status(conflict_id, 'pending', resolution)
return result
def _resolve_duplicate_match(self, conflict: Conflict, resolution: Dict) –> Dict:
"""解决重复匹配冲突"""
action = resolution.get('action')
if action == 'cancel_duplicate':
# 取消重复的匹配
duplicate_match_ids = resolution.get('match_ids', [])
for match_id in duplicate_match_ids:
self.cancel_match(match_id)
# 验证解决效果
trade_id = conflict.resource_id
matches = self.get_matches_by_trade(trade_id)
total_matched = sum(m.quantity for m in matches)
trade = self.get_trade(trade_id)
if total_matched <= trade.quantity:
return {"success": True, "message": "冲突已解决"}
else:
return {"success": False, "reason": "仍有重复匹配"}
elif action == 'adjust_quantity':
# 调整匹配数量
adjustments = resolution.get('adjustments', {})
for match_id, new_quantity in adjustments.items():
self.adjust_match_quantity(match_id, new_quantity)
return {"success": True, "message": "匹配数量已调整"}
return {"success": False, "reason": "未知解决方案"}
def _resolve_logic_inconsistency(self, conflict: Conflict, resolution: Dict) –> Dict:
"""解决逻辑不一致冲突"""
# 触发数据一致性重算
resource_id = conflict.resource_id
if conflict.resource_type == 'contract':
# 重新计算合同执行数量
self.recalculate_contract_quantity(resource_id)
# 验证解决效果
contract = self.get_contract(resource_id)
if contract.linked_quantity <= contract.total_quantity:
return {"success": True, "message": "逻辑已修正"}
else:
return {"success": False, "reason": "逻辑仍不一致，需要人工处理"}
冲突解决支持自动解决与人工解决两种模式，简单冲突可自动解决，复杂冲突需要人工介入。
四、冲突预防机制
除了冲突检测与处理，系统还需要建立冲突预防机制，减少冲突发生。预防措施包括：操作锁定（操作前锁定资源）、乐观锁（版本号控制）、操作队列（串行化操作）、权限控制（防止越权操作）。
class ConflictPrevention:
"""冲突预防"""
def __init__(self):
self.lock_manager = LockManager()
self.operation_queue = OperationQueue()
def prevent_concurrent_modify(self, resource_id: str, user_id: str,
timeout: int = 300) –> bool:
"""防止并发修改"""
# 尝试获取锁
lock_acquired = self.lock_manager.acquire_lock(
resource_id, user_id, timeout=timeout
)
if not lock_acquired:
# 锁获取失败，提示用户
lock_info = self.lock_manager.get_lock(resource_id)
raise ConcurrentModifyError(
f"资源已被用户 {lock_info['user_id']} 锁定，请稍后重试"
)
return True
def queue_operation(self, operation: Dict):
"""将操作加入队列（串行化）"""
resource_id = operation.get('resource_id')
# 检查是否有相同资源的操作在队列中
if self.operation_queue.has_pending_operation(resource_id):
# 加入队列等待
self.operation_queue.enqueue(operation)
else:
# 直接执行
self.operation_queue.execute(operation)
冲突预防机制可显著减少冲突发生，提升系统稳定性。
总结
基差风险管理系统的冲突预警模块是保障数据一致性的重要机制。通过完善的冲突检测、及时的预警通知、规范的处理流程与有效的预防措施，企业可及时发现并处理数据冲突，将冲突发生率降低80%以上，数据一致性提升至99.9%以上。如需了解快期-匹配宝的冲突预警配置方案，可参考相关产品文档。
