Mở Đầu: Khi Dữ Liệu Quá Khứ Trở Thành Ác Mộng
Tôi vẫn nhớ rõ buổi sáng tháng 3 năm 2026, team quant của tôi phát hiện một con số không tưởng: chiến lược giao dịch backtest trên 3 tháng dữ liệu Binance orderbook cho kết quả lợi nhuận 340% — nhưng khi deploy lên live, Drawdown lên tới 67% chỉ trong 2 tuần. Nguyên nhân? Chúng tôi đã vô tình sử dụng dữ liệu orderbook đã qua "làm sạch" bởi một data vendor không rõ nguồn gốc, với bid-ask spread được điều chỉnh sai lệch 2.3 pip so với thực tế.
Đó là khoảnh khắc tôi nhận ra: Lineage audit cho historical market data không phải là optional — nó là survival kit cho bất kỳ ai đang xây dựng hệ thống giao dịch định lượng. Bài viết này sẽ hướng dẫn bạn cách tôi xây dựng hệ thống Tardis lineage tracking từ con số 0, giải quyết bài toán: "Tôi đang dùng data snapshot nào, nó đã qua xử lý gì, và tham số replay được thiết lập ra sao?"
1. Tại Sao Tardis Lineage Audit Lại Quan Trọng?
Trong hệ sinh thái Binance market data, Tardis (bởi Tardis.dev) là một trong những nguồn cung cấp historical orderbook và trade data phổ biến nhất. Tuy nhiên, khi bạn mua data từ nhiều vendor hoặc tự crawl, vấn đề data provenance trở nên cực kỳ phức tạp:
- Data source ambiguity: Orderbook snapshot có thể đến từ Binance API gốc, Binance historical data dumps, hoặc third-party aggregator
- Cleaning version confusion: Mỗi lần vendor "làm sạch" data (loại bỏ outliers, fill gaps, normalize timestamp), một phiên bản mới được tạo nhưng không always được track
- Replay parameter mismatch: Tham số như snapshot interval, orderbook depth, update frequency ảnh hưởng trực tiếp đến backtest accuracy
- Compliance risk: MiFID II và các regulation khác yêu cầu prove data lineage cho audit
2. Kiến Trúc Tardis Lineage Tracking System
2.1 Data Lineage Model
Mô hình lineage của tôi gồm 4 layer chính:
- Source Layer: Raw data từ Binance, được hash và timestamp
- Cleaning Layer: Các transformation steps (gap-fill, outlier removal, normalization)
- Aggregation Layer: Orderbook snapshots ở various intervals
- Replay Layer: Parameters cho simulation và backtesting
2.2 Core Data Structures
"""
Tardis Lineage Data Models
Mô hình lineage cho Binance orderbook snapshots
"""
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
from datetime import datetime
from enum import Enum
import hashlib
import json
class DataSource(Enum):
BINANCE_API = "binance_api"
BINANCE_HISTORY_DUMP = "binance_history_dump"
TARDIS_RAW = "tardis_raw"
TARDIS_CLEANED = "tardis_cleaned"
THIRD_PARTY = "third_party"
class CleaningVersion(Enum):
RAW = "raw"
V1_GAP_FILL = "v1_gap_fill"
V2_OUTLIER_REMOVE = "v2_outlier_remove"
V3_NORMALIZE = "v3_normalize"
V4_FINAL = "v4_final"
@dataclass
class SnapshotMetadata:
"""Metadata cho một orderbook snapshot"""
snapshot_id: str
symbol: str
exchange: str = "binance"
source: DataSource = DataSource.BINANCE_API
timestamp: datetime = field(default_factory=datetime.utcnow)
# Lineage information
cleaning_version: CleaningVersion = CleaningVersion.RAW
cleaning_steps: List[str] = field(default_factory=list)
cleaning_timestamp: Optional[datetime] = None
cleaning_source: Optional[str] = None # API endpoint hoặc vendor name
# Content hash để verify integrity
content_hash: str = ""
previous_snapshot_id: Optional[str] = None
# Replay parameters
replay_interval_ms: int = 100
orderbook_depth: int = 20
update_frequency_hz: int = 10
# Additional metadata
raw_size_bytes: int = 0
compressed_size_bytes: int = 0
compression_algorithm: str = "none"
def compute_hash(self, content: bytes) -> str:
"""Compute SHA-256 hash của snapshot content"""
return hashlib.sha256(content).hexdigest()
def to_dict(self) -> Dict[str, Any]:
"""Export metadata thành dictionary cho storage"""
return {
"snapshot_id": self.snapshot_id,
"symbol": self.symbol,
"exchange": self.exchange,
"source": self.source.value,
"timestamp": self.timestamp.isoformat(),
"cleaning_version": self.cleaning_version.value,
"cleaning_steps": self.cleaning_steps,
"cleaning_timestamp": self.cleaning_timestamp.isoformat() if self.cleaning_timestamp else None,
"cleaning_source": self.cleaning_source,
"content_hash": self.content_hash,
"previous_snapshot_id": self.previous_snapshot_id,
"replay_interval_ms": self.replay_interval_ms,
"orderbook_depth": self.orderbook_depth,
"update_frequency_hz": self.update_frequency_hz,
"raw_size_bytes": self.raw_size_bytes,
"compressed_size_bytes": self.compressed_size_bytes,
"compression_algorithm": self.compression_algorithm,
}
@dataclass
class LineageChain:
"""Complete lineage chain cho một dataset"""
chain_id: str
snapshots: List[SnapshotMetadata] = field(default_factory=list)
created_at: datetime = field(default_factory=datetime.utcnow)
owner: str = ""
purpose: str = "" # backtest, live_trading, research, etc.
def add_snapshot(self, snapshot: SnapshotMetadata):
"""Thêm snapshot vào chain, tự động link previous"""
if self.snapshots:
snapshot.previous_snapshot_id = self.snapshots[-1].snapshot_id
self.snapshots.append(snapshot)
def verify_integrity(self) -> bool:
"""Verify toàn bộ chain integrity"""
for i, snapshot in enumerate(self.snapshots):
if snapshot.previous_snapshot_id:
if i > 0:
prev = self.snapshots[i - 1]
if snapshot.previous_snapshot_id != prev.snapshot_id:
return False
return True
def get_cleaning_summary(self) -> Dict[str, int]:
"""Summary số lượng mỗi cleaning version trong chain"""
summary = {}
for snap in self.snapshots:
version = snap.cleaning_version.value
summary[version] = summary.get(version, 0) + 1
return summary3. Triển Khai Hệ Thống Tracking với HolySheep AI
Để xử lý lineage data một cách intelligent, tôi sử dụng HolySheep AI với chi phí chỉ từ $0.42/MTok (DeepSeek V3.2) — rẻ hơn 95% so với OpenAI. Dưới đây là cách tôi tích hợp:
3.1 Lineage Analyzer Agent
"""
Tardis Lineage Analyzer - Sử dụng HolySheep AI API
Auto-detect data anomalies và cleaning version mismatches
"""
import requests
import json
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from datetime import datetime
========== CẤU HÌNH HOLYSHEEP ==========
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Thay thế bằng API key của bạn
class TardisLineageAnalyzer:
"""
Analyzer sử dụng AI để track và verify Tardis orderbook lineage
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def analyze_snapshot_quality(
self,
snapshot_data: Dict[str, Any]
) -> Dict[str, Any]:
"""
Phân tích quality của một orderbook snapshot
Returns: Quality report với anomaly detection
"""
endpoint = f"{self.base_url}/chat/completions"
prompt = f"""Bạn là một data quality engineer chuyên về cryptocurrency market data.
Phân tích orderbook snapshot sau và trả về JSON:
{{
"quality_score": 0-100,
"anomalies": ["list các anomalies found"],
"bid_ask_spread_pct": spread percentage,
"order_imbalance": -1 đến 1,
"suggested_cleaning_steps": ["list recommended cleaning"],
"is_suspicious": true/false,
"suspicion_reason": "reason nếu suspicious"
}}
Snapshot data:
{json.dumps(snapshot_data, indent=2)}
Chỉ trả về JSON, không có markdown formatting.
"""
payload = {
"model": "deepseek-v3.2", # $0.42/MTok - tiết kiệm 95%
"messages": [
{"role": "system", "content": "You are a data quality expert."},
{"role": "user", "content": prompt}
],
"temperature": 0.1,
"max_tokens": 500
}
response = requests.post(
endpoint,
headers=self.headers,
json=payload,
timeout=30
)
if response.status_code != 200:
raise Exception(f"API Error: {response.status_code} - {response.text}")
result = response.json()
content = result["choices"][0]["message"]["content"]
# Parse JSON từ response
try:
return json.loads(content)
except json.JSONDecodeError:
# Fallback: extract JSON từ response
start = content.find('{')
end = content.rfind('}') + 1
return json.loads(content[start:end])
def detect_lineage_discontinuity(
self,
chain: List[Dict[str, Any]]
) -> Dict[str, Any]:
"""
Detect lineage discontinuities trong một chain snapshots
Phát hiện các trường hợp:
- Timestamp gaps > expected
- Cleaning version jumps không smooth
- Hash mismatches
"""
endpoint = f"{self.base_url}/chat/completions"
prompt = f"""Phân tích lineage chain sau để tìm discontinuities và anomalies.
Chain length: {len(chain)} snapshots
Expected interval: 100ms
Với mỗi snapshot, kiểm tra:
1. Timestamp gap so với previous
2. Cleaning version transition
3. Orderbook depth changes
4. Bid-ask spread anomalies
Trả về JSON:
{{
"discontinuities": [
{{
"position": index,
"snapshot_id": id,
"type": "timestamp_gap|version_jump|depth_change|spread_anomaly",
"severity": "low|medium|high|critical",
"description": "mô tả chi tiết",
"impact_on_backtest": "mô tả impact"
}}
],
"overall_health_score": 0-100,
"recommendation": "recommendation string"
}}
Data:
{json.dumps(chain[:50], indent=2)} # Limit để tránh token overflow
"""
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "system", "content": "You are a blockchain/data lineage expert."},
{"role": "user", "content": prompt}
],
"temperature": 0.1,
"max_tokens": 800
}
response = requests.post(endpoint, headers=self.headers, json=payload, timeout=45)
result = response.json()
return json.loads(result["choices"][0]["message"]["content"])
def generate_lineage_report(
self,
symbol: str,
date_range: tuple,
lineage_chain: Any
) -> str:
"""
Generate comprehensive lineage audit report
"""
endpoint = f"{self.base_url}/chat/completions"
summary = lineage_chain.get_cleaning_summary()
total_snapshots = len(lineage_chain.snapshots)
prompt = f"""Tạo lineage audit report cho:
Symbol: {symbol}
Date range: {date_range[0]} đến {date_range[1]}
Total snapshots: {total_snapshots}
Cleaning version distribution:
{json.dumps(summary, indent=2)}
Integrity verified: {lineage_chain.verify_integrity()}
Report cần bao gồm:
1. Executive Summary
2. Data Quality Metrics
3. Lineage Chain Analysis
4. Risk Assessment
5. Recommendations
Format: Markdown
"""
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "system", "content": "You are a senior data auditor."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 1500
}
response = requests.post(endpoint, headers=self.headers, json=payload, timeout=60)
return response.json()["choices"][0]["message"]["content"]
========== VÍ DỤ SỬ DỤNG ==========
if __name__ == "__main__":
analyzer = TardisLineageAnalyzer(HOLYSHEEP_API_KEY)
# Test với sample orderbook snapshot
sample_snapshot = {
"symbol": "BTCUSDT",
"timestamp": "2026-03-15T10:30:00.123Z",
"bids": [
{"price": 67450.50, "quantity": 1.234},
{"price": 67449.80, "quantity": 2.567},
{"price": 67448.20, "quantity": 0.890},
],
"asks": [
{"price": 67451.00, "quantity": 1.500},
{"price": 67452.30, "quantity": 3.210},
{"price": 67453.80, "quantity": 0.450},
],
"source": "tardis_cached",
"cleaning_version": "v2_outlier_remove"
}
try:
quality_report = analyzer.analyze_snapshot_quality(sample_snapshot)
print(f"Quality Score: {quality_report.get('quality_score')}")
print(f"Anomalies: {quality_report.get('anomalies', [])}")
except Exception as e:
print(f"Error: {e}")3.2 Automated Lineage Verification Pipeline
"""
Tardis Lineage Verification Pipeline
Tự động verify và audit lineage cho historical Binance data
"""
import requests
import hashlib
import json
from datetime import datetime, timedelta
from typing import List, Dict, Optional, Tuple
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class LineageVerificationPipeline:
"""
Pipeline tự động verify Tardis lineage data
Features:
- Hash verification cho data integrity
- Gap detection trong timestamp series
- Cross-reference với Binance official data
- Anomaly detection sử dụng AI
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.base_url = HOLYSHEEP_BASE_URL
def verify_data_integrity(
self,
snapshots: List[Dict]
) -> Dict[str, any]:
"""
Verify data integrity bằng cách:
1. Check hash continuity
2. Detect timestamp gaps
3. Validate data format
"""
verification_result = {
"total_snapshots": len(snapshots),
"hash_breaks": [],
"timestamp_gaps": [],
"format_errors": [],
"integrity_score": 100.0
}
for i in range(len(snapshots) - 1):
current = snapshots[i]
next_snap = snapshots[i + 1]
# Check 1: Timestamp gap detection
try:
curr_ts = datetime.fromisoformat(current["timestamp"].replace("Z", "+00:00"))
next_ts = datetime.fromisoformat(next_snap["timestamp"].replace("Z", "+00:00"))
gap_ms = (next_ts - curr_ts).total_seconds() * 1000
expected_gap = current.get("expected_interval_ms", 100)
if abs(gap_ms - expected_gap) > 50: # >50ms deviation
verification_result["timestamp_gaps"].append({
"position": i,
"expected_ms": expected_gap,
"actual_ms": gap_ms,
"deviation_ms": gap_ms - expected_gap
})
except Exception as e:
verification_result["format_errors"].append({
"position": i,
"error": str(e)
})
# Check 2: Content hash verification (nếu có stored hash)
if "content_hash" in current and "content_hash" in next_snap:
# Verify hash chain integrity
if current["content_hash"] == next_snap.get("previous_hash"):
pass # Hash chain intact
else:
verification_result["hash_breaks"].append({
"position": i,
"expected_hash": next_snap.get("previous_hash"),
"actual_hash": current["content_hash"]
})
# Calculate integrity score
total_issues = (
len(verification_result["hash_breaks"]) +
len(verification_result["timestamp_gaps"]) +
len(verification_result["format_errors"])
)
if len(snapshots) > 0:
verification_result["integrity_score"] = max(
0,
100 - (total_issues / len(snapshots) * 100)
)
return verification_result
def ai_powered_lineage_audit(
self,
symbol: str,
start_date: str,
end_date: str,
lineage_data: List[Dict]
) -> Dict[str, any]:
"""
Sử dụng AI để perform deep lineage audit
Detects subtle anomalies mà rule-based systems miss
"""
endpoint = f"{self.base_url}/chat/completions"
# Prepare summary statistics
timestamps = [datetime.fromisoformat(s["timestamp"].replace("Z", "+00:00"))
for s in lineage_data if "timestamp" in s]
stats = {
"symbol": symbol,
"date_range": f"{start_date} to {end_date}",
"total_snapshots": len(lineage_data),
"first_timestamp": timestamps[0].isoformat() if timestamps else None,
"last_timestamp": timestamps[-1].isoformat() if timestamps else None,
"unique_cleaning_versions": list(set(s.get("cleaning_version", "unknown")
for s in lineage_data)),
"sources": list(set(s.get("source", "unknown") for s in lineage_data)),
}
prompt = f"""Bạn là Senior Data Lineage Auditor chuyên về cryptocurrency market data.
Thực hiện lineage audit cho dataset sau:
Symbol: {symbol}
Date Range: {start_date} to {end_date}
Total Snapshots: {len(lineage_data)}
Cleaning Versions Found: {stats['unique_cleaning_versions']}
Data Sources Found: {stats['sources']}
Kiểm tra và trả về JSON report với cấu trúc:
{{
"audit_timestamp": "{datetime.utcnow().isoformat()}",
"findings": [
{{
"category": "data_quality|lineage_gap|version_inconsistency|source_mix",
"severity": "info|warning|critical",
"description": "mô tả chi tiết",
"affected_snapshots": số lượng,
"recommendation": "hành động cần thiết"
}}
],
"overall_assessment": "pass|warning|fail",
"regulatory_compliance": {{
"miFID_ii_compliant": true/false,
"audit_trail_complete": true/false,
"gaps_identified": []
}},
"estimated_fix_effort_hours": số giờ ước tính
}}
Data sample (first 20 snapshots):
{json.dumps(lineage_data[:20], indent=2)}
"""
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "system", "content": "You are a senior data lineage auditor with expertise in financial market data compliance."},
{"role": "user", "content": prompt}
],
"temperature": 0.1,
"max_tokens": 1200
}
start_time = time.time()
response = requests.post(endpoint, headers=self.headers, json=payload, timeout=90)
latency_ms = (time.time() - start_time) * 1000
print(f"AI Audit completed in {latency_ms:.0f}ms")
result = response.json()
return {
"audit_report": json.loads(result["choices"][0]["message"]["content"]),
"latency_ms": latency_ms,
"cost_estimate": f"${(latency_ms / 1000 * 0.42 / 1000):.4f}" # DeepSeek V3.2 pricing
}
def generate_compliance_certificate(
self,
audit_result: Dict,
symbol: str,
date_range: Tuple[str, str]
) -> str:
"""
Generate compliance certificate cho regulatory purposes
"""
endpoint = f"{self.base_url}/chat/completions"
prompt = f"""Tạo formal compliance certificate cho lineage audit:
Symbol: {symbol}
Date Range: {date_range[0]} to {date_range[1]}
Audit Result Summary:
- Overall Assessment: {audit_result.get('audit_report', {}).get('overall_assessment', 'N/A')}
- Findings Count: {len(audit_result.get('audit_report', {}).get('findings', []))}
- Regulatory Compliance: {json.dumps(audit_result.get('audit_report', {}).get('regulatory_compliance', {}))}
Format: Formal certificate với:
1. Certificate Header
2. Data Scope Declaration
3. Audit Methodology
4. Findings Summary
5. Compliance Statement
6. Digital Signature Placeholder
7. Disclaimer
Language: Tiếng Việt cho phần heading, English cho technical content
"""
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "system", "content": "You are a compliance document generator."},
{"role": "user", "content": prompt}
],
"temperature": 0.2,
"max_tokens": 1000
}
response = requests.post(endpoint, headers=self.headers, json=payload, timeout=60)
return response.json()["choices"][0]["message"]["content"]
========== DEMO USAGE ==========
if __name__ == "__main__":
pipeline = LineageVerificationPipeline(HOLYSHEEP_API_KEY)
# Simulated lineage data
sample_data = [
{
"snapshot_id": f"snap_{i:05d}",
"timestamp": f"2026-03-15T10:30:00.{i:03d}Z",
"source": "tardis_raw" if i % 3 == 0 else "tardis_cleaned",
"cleaning_version": f"v{i % 4}" if i % 3 != 0 else "raw",
"content_hash": f"hash_{i}",
"previous_hash": f"hash_{i-1}" if i > 0 else "genesis"
}
for i in range(100)
]
# Verify integrity
result = pipeline.verify_data_integrity(sample_data)
print(f"Integrity Score: {result['integrity_score']:.1f}%")
print(f"Hash Breaks: {len(result['hash_breaks'])}")
print(f"Timestamp Gaps: {len(result['timestamp_gaps'])}")
# Run AI audit (uncomment để test thực)
# audit_result = pipeline.ai_powered_lineage_audit(
# symbol="BTCUSDT",
# start_date="2026-03-01",
# end_date="2026-03-15",
# lineage_data=sample_data
# )
# print(json.dumps(audit_result, indent=2))4. Replay Parameters Configuration
Một trong những khía cạnh quan trọng nhất của lineage audit là hiểu và track replay parameters. Sai lệch tham số replay có thể biến một backtest tưởng chừng hoàn hảo thành thảm họa live trading.
4.1 Critical Replay Parameters
| Parameter | Typical Values | Impact on Backtest | Risk if Misconfigured |
|---|---|---|---|
| snapshot_interval_ms | 10, 50, 100, 500, 1000 | Độ phân giải thời gian | Miss short-term signals, false precision |
| orderbook_depth | 10, 20, 50, 100, 500 | Liquidity model accuracy | Over/under estimate slippage |
| update_frequency_hz | 1, 5, 10, 100 | Event detection speed | Miss micro-structure events |
| latency_simulation_ms | 0, 50, 100, 200 | Realism of execution | Unrealistic fills, false alpha |
| fee_tier | taker: 0.04%, maker: 0.02% | Net P&L calculation | Inflated returns by 2-5% |
4.2 Tardis Replay Configuration Manager
"""
Tardis Replay Configuration Manager
Quản lý và track tất cả replay parameters cho lineage
"""
from dataclasses import dataclass, field, asdict
from typing import Dict, List, Optional, Any
from datetime import datetime
import json
import hashlib
@dataclass
class ReplayParameters:
"""Tất cả parameters cho orderbook replay"""
# Core timing
snapshot_interval_ms: int = 100
warmup_duration_ms: int = 1000 # Chờ orderbook stable trước khi replay
cooldown_duration_ms: int = 500
# Orderbook config
orderbook_depth: int = 20 # Số levels mỗi side
include_aggregated: bool = True
include_order_updates: bool = True
# Simulation parameters
latency_simulation_ms: int = 0
slippage_model: str = "fixed" # fixed, variable, volume_based
slippage_bps: float = 0.5
# Fee structure (Binance VIP 0)
maker_fee_bps: float = 0.02
taker_fee_bps: float = 0.04
# Data source specific
source_exchange: str = "binance"
data_format: str = "tardis_json" # tardis_json, binance_compressed, raw_csv
# Validation
max_gap_allowed_ms: int = 5000
fill_missing_with: str = "interpolation" # interpolation, previous, none
def compute_config_hash(self) -> str:
"""Compute unique hash cho configuration"""
config_str = json.dumps(asdict(self), sort_keys=True)
return hashlib.sha256(config_str.encode()).hexdigest()[:16]
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "ReplayParameters":
return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})
class ReplayParameterHistory:
"""
Track history của tất cả replay parameter changes
Phục vụ lineage audit và reproducibility
"""
def __init__(self, experiment_id: str):
self.experiment_id = experiment_id
self.changes: List[Dict[str, Any]] = []
self.current_config: Optional[ReplayParameters] = None
def set_parameters(
self,
params: ReplayParameters,
reason: str,
changed_by: str = "system"
):
"""Ghi nhận parameter change vào history"""
timestamp = datetime.utcnow()
change_record = {
"timestamp": timestamp.isoformat(),
"experiment_id": self.experiment_id,
"reason": reason,
"changed_by": changed_by,
"config_hash": params.compute_config_hash(),
"parameters": params.to_dict(),
"change_id": len(self.changes)
}
# Nếu có config trước, record diff
if self.current_config:
change_record["diff"] = self._compute_diff(
self.current_config,
params
)
self.changes.append(change_record)
self.current_config = params
def _compute_diff(
self,
old: ReplayParameters,
new: ReplayParameters
) -> Dict[str, Dict]:
"""Compute diff giữa 2 configurations"""
old_dict = old.to_dict()
new_dict = new.to_dict()
diff = {}
for key in new_dict:
if old_dict.get(key) != new_dict[key]:
diff[key] = {
"old": old_dict[key],
"new": new_dict[key]
}
return diff
def get_config_at_timestamp(
self,
timestamp: datetime
) -> Optional[ReplayParameters]:
"""Lấy configuration tại một thời điểm cụ thể"""
for change in reversed(self.changes):
change_time = datetime.fromisoformat(change["timestamp"])
if change_time <= timestamp:
return ReplayParameters.from_dict(change["parameters"])
return None
def generate_replay_manifest(self) -> Dict[str, Any]:
"""Generate manifest cho reproducibility"""
if not self.current_config:
return {"error": "No configuration set"}
return {
"experiment_id": self.experiment_id,
"manifest_version": "1.0",
"generated_at": datetime.utcnow().isoformat(),
"total_changes": len(self.changes),
"final_config_hash": self.current_config.compute_config_hash(),
"config_timeline": [
{
"change_id": c["change_id"],
"timestamp": c["timestamp"],
"reason": c["reason"],
"config_hash": c["config_hash"]
}
for c in self.changes
],
"final_parameters": self
Tài nguyên liên quan
Bài viết liên quan