在期现业务管理中，期现对账是确保期货成交与现货合同匹配准确性的重要环节。基差风险管理系统的期现对账流程，通过系统化的对账规则与自动化的校验机制，实现了高效、准确的期现数据核对。本文将详细说明对账流程的设计思路、校验规则与异常处理。
一、对账流程的核心步骤
期现对账流程包括数据准备、数据匹配、差异识别、差异分析、差异处理五个核心步骤。数据准备阶段收集期货成交数据与现货合同数据，确保数据完整。数据匹配阶段将成交与合同建立关联关系。差异识别阶段找出不匹配的数据。差异分析阶段分析差异原因。差异处理阶段修正差异数据。
在快期-匹配宝系统中，对账流程支持自动执行与手动触发两种模式，自动对账每日定时执行，手动对账支持按需执行。
from dataclasses import dataclass
from typing import List, Dict
from datetime import datetime, date
@dataclass
class ReconciliationTask:
"""对账任务"""
task_id: str
reconciliation_date: date
account_id: str = None
status: str = "pending" # pending, running, completed, failed
created_at: datetime = None
class ReconciliationEngine:
"""对账引擎"""
def __init__(self):
self.trade_provider = TradeDataProvider()
self.contract_provider = ContractDataProvider()
self.match_validator = MatchValidator()
def execute_reconciliation(self, task: ReconciliationTask) –> Dict:
"""执行对账"""
try:
# 1. 数据准备
print(f"[对账] 开始数据准备…")
trade_data = self._prepare_trade_data(task)
contract_data = self._prepare_contract_data(task)
# 2. 数据匹配
print(f"[对账] 开始数据匹配…")
matches = self._match_trades_to_contracts(trade_data, contract_data)
# 3. 差异识别
print(f"[对账] 开始差异识别…")
differences = self._identify_differences(trade_data, contract_data, matches)
# 4. 差异分析
print(f"[对账] 开始差异分析…")
analysis = self._analyze_differences(differences)
# 5. 生成对账报告
report = self._generate_reconciliation_report(task, matches, differences, analysis)
# 6. 更新任务状态
task.status = "completed"
return {
"task_id": task.task_id,
"status": "completed",
"total_trades": len(trade_data),
"total_contracts": len(contract_data),
"matched_count": len(matches),
"difference_count": len(differences),
"report": report
}
except Exception as e:
task.status = "failed"
return {
"task_id": task.task_id,
"status": "failed",
"error": str(e)
}
def _prepare_trade_data(self, task: ReconciliationTask) –> List[Dict]:
"""准备期货成交数据"""
filters = {
"date": task.reconciliation_date
}
if task.account_id:
filters["account_id"] = task.account_id
trades = self.trade_provider.get_trades(filters)
return [{
"trade_id": t.id,
"symbol": t.symbol,
"direction": t.direction,
"quantity": t.quantity,
"price": t.price,
"trade_date": t.trade_date,
"account_id": t.account_id
} for t in trades]
def _prepare_contract_data(self, task: ReconciliationTask) –> List[Dict]:
"""准备现货合同数据"""
filters = {
"active_on_date": task.reconciliation_date
}
if task.account_id:
filters["account_id"] = task.account_id
contracts = self.contract_provider.get_contracts(filters)
return [{
"contract_id": c.id,
"commodity": c.commodity,
"type": c.type,
"total_quantity": c.total_quantity,
"linked_quantity": c.linked_quantity,
"remaining_quantity": c.total_quantity – c.linked_quantity,
"customer_id": c.customer_id
} for c in contracts]
def _match_trades_to_contracts(self, trades: List[Dict],
contracts: List[Dict]) –> List[Dict]:
"""匹配成交与合同"""
matches = []
for trade in trades:
# 查找匹配的合同
matching_contracts = self._find_matching_contracts(trade, contracts)
for contract in matching_contracts:
# 检查是否已匹配
existing_match = self.match_validator.get_match(trade['trade_id'], contract['contract_id'])
if existing_match:
matches.append(existing_match)
else:
# 创建新匹配
match = self._create_match(trade, contract)
matches.append(match)
return matches
对账流程支持断点续传，中断后可从中断点继续执行。
二、对账规则的配置
对账规则决定如何匹配成交与合同，规则包括：品种匹配（期货品种与合同品种一致）、方向匹配（买入成交匹配采购合同，卖出成交匹配销售合同）、时间匹配（成交时间在合同有效期内）、数量匹配（匹配数量不超过合同剩余数量）。
class ReconciliationRule:
"""对账规则"""
def __init__(self, rule_config: Dict):
self.rule_id = rule_config['rule_id']
self.rule_name = rule_config['rule_name']
self.match_criteria = rule_config['match_criteria']
self.priority = rule_config.get('priority', 10)
self.enabled = rule_config.get('enabled', True)
def evaluate(self, trade: Dict, contract: Dict) –> bool:
"""评估是否匹配"""
criteria = self.match_criteria
# 品种匹配
if 'commodity_match' in criteria and criteria['commodity_match']:
if trade['symbol'][:2] != contract['commodity']:
return False
# 方向匹配
if 'direction_match' in criteria and criteria['direction_match']:
expected_direction = 'buy' if contract['type'] == 'purchase' else 'sell'
if trade['direction'] != expected_direction:
return False
# 时间匹配
if 'time_match' in criteria and criteria['time_match']:
if not (contract.get('start_date') <= trade['trade_date'] <= contract.get('end_date')):
return False
# 数量匹配
if 'quantity_match' in criteria and criteria['quantity_match']:
if trade['quantity'] > contract['remaining_quantity']:
return False
return True
class ReconciliationRuleEngine:
"""对账规则引擎"""
def __init__(self):
self.rules = []
def add_rule(self, rule: ReconciliationRule):
"""添加对账规则"""
self.rules.append(rule)
# 按优先级排序
self.rules.sort(key=lambda r: r.priority, reverse=True)
def find_matching_contract(self, trade: Dict, contracts: List[Dict]) –> Dict:
"""查找匹配的合同"""
for rule in self.rules:
if not rule.enabled:
continue
for contract in contracts:
if rule.evaluate(trade, contract):
return contract
return None
对账规则支持灵活配置，用户可根据业务需求自定义匹配规则。
三、差异分析与处理
对账过程中发现的差异需要进行分析与处理。差异类型包括：未匹配成交（成交未匹配到合同）、未匹配合同（合同未匹配到成交）、数量差异（匹配数量与预期不一致）、价格差异（成交价格与合同价格偏差大）。
class DifferenceAnalyzer:
"""差异分析器"""
def analyze_differences(self, differences: List[Dict]) –> Dict:
"""分析差异"""
analysis = {
"total_differences": len(differences),
"by_type": {},
"by_severity": {},
"recommendations": []
}
# 按类型分类
for diff in differences:
diff_type = diff['type']
if diff_type not in analysis['by_type']:
analysis['by_type'][diff_type] = []
analysis['by_type'][diff_type].append(diff)
# 按严重程度分类
for diff in differences:
severity = self._assess_severity(diff)
if severity not in analysis['by_severity']:
analysis['by_severity'][severity] = []
analysis['by_severity'][severity].append(diff)
# 生成处理建议
analysis['recommendations'] = self._generate_recommendations(differences)
return analysis
def _assess_severity(self, difference: Dict) –> str:
"""评估差异严重程度"""
diff_type = difference['type']
if diff_type == 'unmatched_trade_large_quantity':
return 'high'
elif diff_type == 'quantity_mismatch':
return 'medium'
elif diff_type == 'price_deviation':
# 价格偏差小于1%为低严重程度
if difference.get('deviation_rate', 0) < 0.01:
return 'low'
else:
return 'medium'
else:
return 'low'
def _generate_recommendations(self, differences: List[Dict]) –> List[str]:
"""生成处理建议"""
recommendations = []
# 统计未匹配成交
unmatched_trades = [d for d in differences if d['type'] == 'unmatched_trade']
if len(unmatched_trades) > 10:
recommendations.append(f"发现{len(unmatched_trades)}笔未匹配成交，建议检查匹配规则配置")
# 统计数量差异
quantity_diffs = [d for d in differences if d['type'] == 'quantity_mismatch']
if quantity_diffs:
total_diff = sum(abs(d.get('difference', 0)) for d in quantity_diffs)
recommendations.append(f"发现数量差异，总差异{total_diff}，建议核对匹配关系")
return recommendations
class DifferenceResolver:
"""差异解决器"""
def resolve_difference(self, difference: Dict, resolution: Dict) –> Dict:
"""解决差异"""
diff_type = difference['type']
if diff_type == 'unmatched_trade':
# 手动匹配成交到合同
return self._resolve_unmatched_trade(difference, resolution)
elif diff_type == 'quantity_mismatch':
# 调整匹配数量
return self._resolve_quantity_mismatch(difference, resolution)
elif diff_type == 'price_deviation':
# 价格差异通常不需要处理（市场价格波动正常）
return {"success": True, "message": "价格差异在可接受范围内"}
return {"success": False, "reason": "未知差异类型"}
def _resolve_unmatched_trade(self, difference: Dict, resolution: Dict) –> Dict:
"""解决未匹配成交"""
trade_id = difference['trade_id']
contract_id = resolution.get('contract_id')
if not contract_id:
return {"success": False, "reason": "未指定合同ID"}
# 创建匹配关系
match_result = self.create_match(trade_id, contract_id, resolution.get('quantity'))
return match_result
差异处理支持自动处理与人工处理，简单差异可自动处理，复杂差异需要人工确认。
四、对账报告的生成
对账完成后需要生成详细的对账报告，报告内容包括：对账摘要（总体统计）、匹配明细（匹配关系明细）、差异明细（差异数据明细）、处理建议（差异处理建议）。
class ReconciliationReportGenerator:
"""对账报告生成器"""
def generate_report(self, task: ReconciliationTask,
matches: List[Dict],
differences: List[Dict],
analysis: Dict) –> Dict:
"""生成对账报告"""
report = {
"report_id": f"RECON_{task.task_id}",
"reconciliation_date": task.reconciliation_date.strftime('%Y-%m-%d'),
"account_id": task.account_id,
"summary": {
"total_trades": analysis.get('total_trades', 0),
"total_contracts": analysis.get('total_contracts', 0),
"matched_count": len(matches),
"unmatched_trades": len([d for d in differences if d['type'] == 'unmatched_trade']),
"unmatched_contracts": len([d for d in differences if d['type'] == 'unmatched_contract']),
"match_rate": len(matches) / analysis.get('total_trades', 1) * 100 if analysis.get('total_trades', 0) > 0 else 0
},
"matches": matches,
"differences": differences,
"analysis": analysis,
"generated_at": datetime.now().isoformat()
}
return report
def export_to_excel(self, report: Dict, file_path: str):
"""导出为Excel"""
import pandas as pd
with pd.ExcelWriter(file_path, engine='openpyxl') as writer:
# 摘要表
summary_df = pd.DataFrame([report['summary']])
summary_df.to_excel(writer, sheet_name='对账摘要', index=False)
# 匹配明细表
if report['matches']:
matches_df = pd.DataFrame(report['matches'])
matches_df.to_excel(writer, sheet_name='匹配明细', index=False)
# 差异明细表
if report['differences']:
differences_df = pd.DataFrame(report['differences'])
differences_df.to_excel(writer, sheet_name='差异明细', index=False)
# 分析建议表
if report['analysis'].get('recommendations'):
recommendations_df = pd.DataFrame(
[{"建议": rec} for rec in report['analysis']['recommendations']]
)
recommendations_df.to_excel(writer, sheet_name='处理建议', index=False)
对账报告支持导出为Excel、PDF等格式，便于分发与存档。
总结
基差风险管理系统的期现对账流程是确保数据准确性的重要机制。通过规范的对账流程、灵活的对账规则、深入的差异分析与详细的对账报告，企业可实现高效的期现数据核对，将对账时间从数小时缩短至数十分钟，对账准确率提升至99.5%以上。如需了解快期-匹配宝的对账配置方案，可参考相关产品文档。
