Là một data engineer làm việc với dữ liệu thời gian thực, tôi đã từng đối mặt với vô số lỗi khi xử lý data pipeline. Một trong những lỗi đáng nhớ nhất là khi tôi cố gắng tải 50GB dữ liệu từ Tardis API và nhận được liên tục các thông báo lỗi: ConnectionError: timeout after 30000ms, tiếp followed by 429 Too Many Requests. Sau 3 ngày debug, tôi mới tìm ra root cause — không phải do API có vấn đề, mà do cách tôi xử lý data streaming hoàn toàn sai. Bài viết này sẽ chia sẻ những kinh nghiệm thực chiến để bạn tránh lặp lại những sai lầm tương tự.
Tardis API là gì và tại sao nên dùng Parquet?
Tardis API là một dịch vụ cung cấp dữ liệu thị trường tài chính, crypto và các sự kiện kinh tế với độ trễ thấp. Parquet là định dạng columnar storage được thiết kế đặc biệt cho phân tích dữ liệu lớn, có khả năng nén cao (thường đạt 2-4x so với CSV) và hỗ trợ predicate pushdown giúp giảm đáng kể thời gian đọc dữ liệu.
Cài đặt môi trường và dependencies
Trước khi bắt đầu, hãy đảm bảo bạn đã cài đặt các thư viện cần thiết. Tôi khuyến nghị sử dụng Python 3.10+ với virtual environment để tránh xung đột dependency.
# Tạo virtual environment
python -m venv tardis-env
source tardis-env/bin/activate # Linux/Mac
tardis-env\Scripts\activate # Windows
Cài đặt dependencies
pip install requests pandas pyarrow duckdb httpx aiohttp \
python-dotenv s3fs pyarrow parquet-tools
Kiểm tra phiên bản
python -c "import duckdb; print(f'DuckDB version: {duckdb.__version__}')"
Output: DuckDB version: 0.9.2
Tải dữ liệu Parquet từ Tardis API
Method 1: Synchronous Download với Retry Logic
import requests
import pandas as pd
import time
from pathlib import Path
from typing import Optional, List
class TardisParquetDownloader:
"""Downloader với retry logic và progress tracking"""
BASE_URL = "https://api.tardis.io/v1"
CHUNK_SIZE = 8192
MAX_RETRIES = 5
RETRY_DELAY = 2
def __init__(self, api_key: str, output_dir: str = "./data"):
self.api_key = api_key
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Accept": "application/x-parquet",
"User-Agent": "TardisClient/2.1.0"
})
def _make_request(self, url: str, params: dict = None) -> requests.Response:
"""Gửi request với exponential backoff retry"""
for attempt in range(self.MAX_RETRIES):
try:
response = self.session.get(
url,
params=params,
timeout=(10, 60), # (connect, read) timeout
stream=True
)
response.raise_for_status()
return response
except requests.exceptions.RequestException as e:
wait_time = self.RETRY_DELAY * (2 ** attempt)
print(f"Attempt {attempt + 1} failed: {e}")
if attempt < self.MAX_RETRIES - 1:
print(f"Retrying in {wait_time}s...")
time.sleep(wait_time)
else:
raise
raise Exception("Max retries exceeded")
def download_parquet(
self,
dataset: str,
symbols: List[str],
start_date: str,
end_date: str,
filters: Optional[dict] = None
) -> Path:
"""Tải dữ liệu Parquet với filtering"""
url = f"{self.BASE_URL}/download/parquet"
params = {
"dataset": dataset,
"symbols": ",".join(symbols),
"start": start_date,
"end": end_date,
"compression": "snappy",
"useDeprecatedFormats": "false"
}
if filters:
params.update(filters)
filename = f"{dataset}_{start_date}_{end_date}.parquet"
filepath = self.output_dir / filename
print(f"Downloading {filename}...")
response = self._make_request(url, params)
total_size = int(response.headers.get("content-length", 0))
downloaded = 0
with open(filepath, "wb") as f:
for chunk in response.iter_content(chunk_size=self.CHUNK_SIZE):
if chunk:
f.write(chunk)
downloaded += len(chunk)
if total_size:
progress = (downloaded / total_size) * 100
print(f"\rProgress: {progress:.1f}%", end="")
print(f"\nSaved to {filepath}")
return filepath
def download_and_convert(
self,
dataset: str,
symbols: List[str],
start_date: str,
end_date: str
) -> pd.DataFrame:
"""Tải và chuyển đổi sang DataFrame"""
filepath = self.download_parquet(dataset, symbols, start_date, end_date)
df = pd.read_parquet(filepath)
print(f"Loaded {len(df):,} rows, {df.memory_usage(deep=True).sum() / 1024**2:.1f} MB")
return df
Sử dụng
if __name__ == "__main__":
downloader = TardisParquetDownloader(
api_key="YOUR_TARDIS_API_KEY",
output_dir="./market_data"
)
df = downloader.download_and_convert(
dataset="crypto_ohlcv",
symbols=["BTC-USD", "ETH-USD"],
start_date="2024-01-01",
end_date="2024-03-01"
)
print(df.head())Method 2: Async Download với aiohttp cho high throughput
import asyncio
import aiohttp
import aiofiles
from dataclasses import dataclass
from typing import List, Optional
import json
@dataclass
class DownloadTask:
url: str
filepath: str
expected_size: Optional[int] = None
class AsyncTardisDownloader:
"""Async downloader cho batch processing với concurrency control"""
BASE_URL = "https://api.tardis.io/v1"
MAX_CONCURRENT = 5
CHUNK_SIZE = 65536
def __init__(self, api_key: str, output_dir: str = "./data"):
self.api_key = api_key
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.semaphore = asyncio.Semaphore(self.MAX_CONCURRENT)
def _get_headers(self) -> dict:
return {
"Authorization": f"Bearer {self.api_key}",
"Accept": "application/x-parquet"
}
async def _download_file(
self,
session: aiohttp.ClientSession,
task: DownloadTask
) -> dict:
"""Download một file với progress"""
async with self.semaphore:
result = {
"url": task.url,
"filepath": task.filepath,
"status": "pending",
"size": 0,
"error": None
}
try:
async with session.get(task.url, headers=self._get_headers()) as response:
response.raise_for_status()
total_size = int(response.headers.get("content-length", 0))
result["expected_size"] = total_size
async with aiofiles.open(task.filepath, "wb") as f:
downloaded = 0
async for chunk in response.content.iter_chunked(self.CHUNK_SIZE):
await f.write(chunk)
downloaded += len(chunk)
if total_size and downloaded % (1024 * 1024) == 0:
progress = (downloaded / total_size) * 100
print(f"[{task.filepath}] {progress:.0f}%")
result["status"] = "success"
result["size"] = downloaded
except Exception as e:
result["status"] = "error"
result["error"] = str(e)
return result
async def batch_download(self, tasks: List[DownloadTask]) -> List[dict]:
"""Download nhiều file đồng thời"""
connector = aiohttp.TCPConnector(limit=self.MAX_CONCURRENT)
timeout = aiohttp.ClientTimeout(total=3600)
async with aiohttp.ClientSession(
connector=connector,
timeout=timeout
) as session:
results = await asyncio.gather(
*[self._download_file(session, task) for task in tasks],
return_exceptions=True
)
valid_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
valid_results.append({
"url": tasks[i].url,
"status": "error",
"error": str(result)
})
else:
valid_results.append(result)
return valid_results
async def download_with_filters(
self,
symbols: List[str],
start_date: str,
end_date: str,
interval: str = "1h"
) -> List[Path]:
"""Tạo và download multiple files dựa trên symbols"""
tasks = []
for symbol in symbols:
url = (
f"{self.BASE_URL}/download/parquet"
f"?symbol={symbol}&start={start_date}&end={end_date}"
f"&interval={interval}&compression=snappy"
)
filename = f"{symbol.replace('-', '_')}_{interval}_{start_date}_{end_date}.parquet"
filepath = str(self.output_dir / filename)
tasks.append(DownloadTask(url=url, filepath=filepath))
results = await self.batch_download(tasks)
successful = [Path(r["filepath"]) for r in results if r["status"] == "success"]
print(f"\nDownloaded {len(successful)}/{len(tasks)} files successfully")
return successful
Chạy async download
async def main():
downloader = AsyncTardisDownloader(
api_key="YOUR_TARDIS_API_KEY",
output_dir="./market_data"
)
symbols = ["BTC-USD", "ETH-USD", "SOL-USD", "DOGE-USD", "XRP-USD"]
files = await downloader.download_with_filters(
symbols=symbols,
start_date="2024-01-01",
end_date="2024-03-01",
interval="1h"
)
return files
asyncio.run(main())
Tối ưu hóa Query với DuckDB
DuckDB là một embedded OLAP database được thiết kế cho analytical workloads. Khi kết hợp với Parquet files từ Tardis API, DuckDB có thể đạt hiệu suất query vượt trội nhờ vectorized execution và predicate pushdown. Dưới đây là các best practices tôi đã áp dụng trong production.
Setup DuckDB với Parquet Virtual Tables
import duckdb
import pandas as pd
from pathlib import Path
from typing import List, Optional
class TardisDuckDBOptimizer:
"""DuckDB optimizer với caching và query optimization"""
def __init__(self, data_dir: str = "./market_data"):
self.data_dir = Path(data_dir)
self.conn = duckdb.connect(database=":memory:") # In-memory database
# Cấu hình DuckDB settings cho performance
self.conn.execute("SET threads TO 8")
self.conn.execute("SET force_index_join TO false")
self.conn.execute("SET enable_progress_bar TO true")
self.conn.execute("SET enable_progress_bar_print TO true")
def register_parquet_files(self, pattern: str = "*.parquet") -> int:
"""Register tất cả Parquet files trong thư mục"""
parquet_files = list(self.data_dir.glob(pattern))
if not parquet_files:
print(f"No Parquet files found matching: {pattern}")
return 0
for f in parquet_files:
table_name = f.stem.replace("-", "_").replace(" ", "_")
# Sử dụng AUTO_DETECT để DuckDB tự infer schema
self.conn.execute(f"""
CREATE VIEW IF NOT EXISTS {table_name} AS
SELECT * FROM read_parquet('{f}', auto_detect=true)
""")
print(f"Registered {len(parquet_files)} Parquet files")
return len(parquet_files)
def explain_query(self, query: str) -> str:
"""Phân tích query execution plan"""
result = self.conn.execute(f"EXPLAIN ANALYZE {query}").fetchdf()
return result.iloc[0, 0]
def execute_analytics(
self,
start_date: str,
end_date: str,
symbols: Optional[List[str]] = None
) -> pd.DataFrame:
"""Thực hiện các phân tích phổ biến"""
symbol_filter = ""
if symbols:
symbol_list = "', '".join(symbols)
symbol_filter = f"AND symbol IN ('{symbol_list}')"
# Query 1: Tính returns và volatility theo ngày
daily_stats_query = f"""
WITH price_data AS (
SELECT
symbol,
timestamp,
close,
LAG(close) OVER (PARTITION BY symbol ORDER BY timestamp) as prev_close
FROM read_parquet('{self.data_dir}/*.parquet')
WHERE timestamp >= '{start_date}'
AND timestamp < '{end_date}'
{symbol_filter}
),
returns AS (
SELECT
symbol,
DATE_TRUNC('day', timestamp) as date,
close,
(close - prev_close) / prev_close * 100 as return_pct
FROM price_data
WHERE prev_close IS NOT NULL
)
SELECT
symbol,
date,
AVG(return_pct) as avg_return,
STDDEV_SAMP(return_pct) as volatility,
MIN(close) as low,
MAX(close) as high,
COUNT(*) as observations
FROM returns
GROUP BY symbol, date
ORDER BY date DESC, symbol
"""
print("Executing daily statistics query...")
result = self.conn.execute(daily_stats_query).fetchdf()
return result
def create_aggregated_parquet(
self,
output_path: str,
aggregation: str = "hourly",
symbols: Optional[List[str]] = None
):
"""Tạo aggregated Parquet file để tăng tốc truy vấn sau"""
symbol_filter = ""
if symbols:
symbol_list = "', '".join(symbols)
symbol_filter = f"WHERE symbol IN ('{symbol_list}')"
time_trunc = {
"hourly": "hour",
"daily": "day",
"weekly": "week"
}.get(aggregation, "day")
create_query = f"""
COPY (
SELECT
symbol,
DATE_TRUNC('{time_trunc}', timestamp) as timestamp,
FIRST(open) as open,
MAX(high) as high,
MIN(low) as low,
LAST(close) as close,
SUM(volume) as volume,
COUNT(*) as bar_count
FROM read_parquet('{self.data_dir}/*.parquet')
{symbol_filter}
GROUP BY symbol, DATE_TRUNC('{time_trunc}', timestamp)
ORDER BY symbol, timestamp
) TO '{output_path}' (FORMAT PARQUET, COMPRESSION 'zstd', ROW_GROUP_SIZE 100000)
"""
print(f"Creating {aggregation} aggregated file...")
self.conn.execute(create_query)
print(f"Saved to {output_path}")
def close(self):
"""Đóng connection"""
self.conn.close()
Sử dụng optimizer
optimizer = TardisDuckDBOptimizer(data_dir="./market_data")
optimizer.register_parquet_files("crypto_*.parquet")
Thực hiện analytics
stats = optimizer.execute_analytics(
start_date="2024-01-01",
end_date="2024-03-01",
symbols=["BTC-USD", "ETH-USD"]
)
print(stats.head(10))
Tạo aggregated file cho production queries
optimizer.create_aggregated_parquet(
output_path="./market_data/btc_eth_hourly.parquet",
aggregation="hourly",
symbols=["BTC-USD", "ETH-USD"]
)
optimizer.close()Advanced: Parallel Query Execution với Multiple Files
import duckdb
import pandas as pd
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Tuple
import pyarrow.parquet as pq
class ParallelDuckDBQuery:
"""Query execution với parallelism trên nhiều Parquet files"""
def __init__(self, n_workers: int = 4):
self.n_workers = n_workers
self.connections = []
# Tạo nhiều connections cho parallel queries
for _ in range(n_workers):
conn = duckdb.connect(database=":memory:")
conn.execute("SET threads TO 2")
self.connections.append(conn)
def _read_single_file(self, filepath: str) -> pd.DataFrame:
"""Đọc một Parquet file"""
return pd.read_parquet(filepath)
def _process_partition(
self,
files: List[str],
conn_idx: int,
query_template: str
) -> pd.DataFrame:
"""Xử lý một partition của files"""
conn = self.connections[conn_idx]
# Register files vào connection
for i, f in enumerate(files):
table_name = f"t{i}"
conn.execute(f"CREATE TEMP TABLE {table_name} AS SELECT * FROM read_parquet('{f}')")
# Execute query
tables = [f"t{i}" for i in range(len(files))]
combined_query = f"SELECT * FROM ({query_template.format(tables=tables[0])})"
result = conn.execute(combined_query).fetchdf()
# Cleanup
for t in tables:
conn.execute(f"DROP TABLE IF EXISTS {t}")
return result
def parallel_query(
self,
files: List[str],
query_template: str,
combine_func: str = "UNION ALL"
) -> pd.DataFrame:
"""Thực hiện query song song trên nhiều files"""
# Chia files thành partitions
chunk_size = max(1, len(files) // self.n_workers)
partitions = [
files[i:i + chunk_size]
for i in range(0, len(files), chunk_size)
]
print(f"Processing {len(files)} files in {len(partitions)} partitions")
results = []
with ThreadPoolExecutor(max_workers=self.n_workers) as executor:
futures = []
for i, partition in enumerate(partitions):
future = executor.submit(
self._process_partition,
partition,
i % self.n_workers,
query_template
)
futures.append(future)
for future in as_completed(futures):
try:
result = future.result()
results.append(result)
except Exception as e:
print(f"Partition failed: {e}")
# Combine results
if not results:
return pd.DataFrame()
combined = pd.concat(results, ignore_index=True)
print(f"Final result: {len(combined):,} rows")
return combined
def close(self):
for conn in self.connections:
conn.close()
Sử dụng parallel query
files = list(Path("./market_data").glob("*.parquet"))
print(f"Found {len(files)} Parquet files")
parallel = ParallelDuckDBQuery(n_workers=4)
Query tính volume-weighted average price
result = parallel.parallel_query(
files=[str(f) for f in files[:20]], # Test với 20 files đầu
query_template="""
SELECT
symbol,
DATE_TRUNC('hour', timestamp) as hour,
SUM(volume * close) / SUM(volume) as vwap,
SUM(volume) as total_volume
FROM {tables}
WHERE timestamp >= '2024-01-01' AND timestamp < '2024-02-01'
GROUP BY symbol, hour
ORDER BY hour DESC
"""
)
print(result.head())
parallel.close()Lỗi thường gặp và cách khắc phục
1. Lỗi "HTTP 401 Unauthorized" hoặc "403 Forbidden"
# Nguyên nhân: API key không hợp lệ hoặc hết hạn
Cách khắc phục:
import os
from pathlib import Path
def validate_api_key():
"""Kiểm tra và validate API key"""
# 1. Kiểm tra environment variable
api_key = os.environ.get("TARDIS_API_KEY")
# 2. Hoặc đọc từ file .env
if not api_key:
from dotenv import load_dotenv
load_dotenv()
api_key = os.environ.get("TARDIS_API_KEY")
# 3. Hoặc sử dụng key file (an toàn hơn cho production)
key_file = Path.home() / ".config" / "tardis" / "api_key"
if not api_key and key_file.exists():
api_key = key_file.read_text().strip()
if not api_key:
raise ValueError("""
Tardis API key không được tìm thấy.
Vui lòng thiết lập qua một trong các cách sau:
1. Environment variable:
export TARDIS_API_KEY='your-key-here'
2. File .env:
TARDIS_API_KEY='your-key-here'
3. Key file:
~/.config/tardis/api_key
Lấy API key tại: https://api.tardis.io/auth
""")
# 4. Validate format (Tardis key thường có prefix 'ts_')
if not api_key.startswith('ts_'):
print(f"Warning: API key format có thể không đúng. "
f"Expected prefix 'ts_', got '{api_key[:5]}...'")
return api_key
Sử dụng
api_key = validate_api_key()
print(f"API key validated: {api_key[:8]}...")2. Lỗi "ConnectionError: timeout after 30000ms" và "429 Too Many Requests"
# Nguyên nhân: Quá nhiều requests trong thời gian ngắn hoặc network latency cao
Cách khắc phục:
import time
import threading
from collections import deque
from typing import Callable, Any
import functools
class RateLimiter:
"""Token bucket rate limiter với thread safety"""
def __init__(self, requests_per_second: float = 5, burst: int = 10):
self.rate = requests_per_second
self.burst = burst
self.tokens = burst
self.last_update = time.time()
self.lock = threading.Lock()
def acquire(self, blocking: bool = True, timeout: float = 60) -> bool:
"""Acquire a token, optionally blocking"""
deadline = time.time() + timeout
while True:
with self.lock:
# Refill tokens
now = time.time()
elapsed = now - self.last_update
self.tokens = min(self.burst, self.tokens + elapsed * self.rate)
self.last_update = now
if self.tokens >= 1:
self.tokens -= 1
return True
if not blocking:
return False
if time.time() >= deadline:
return False
time.sleep(0.1) # Wait before retrying
class RobustAPIClient:
"""Client với retry logic, rate limiting và circuit breaker"""
def __init__(
self,
api_key: str,
requests_per_second: float = 5,
max_retries: int = 5
):
self.api_key = api_key
self.max_retries = max_retries
self.rate_limiter = RateLimiter(requests_per_second=requests_per_second)
# Circuit breaker state
self.failure_count = 0
self.failure_threshold = 5
self.circuit_open = False
self.circuit_open_time = None
self.circuit_reset_timeout = 60 # seconds
# Request history for monitoring
self.request_history = deque(maxlen=1000)
def _should_retry(self, error: Exception, attempt: int) -> bool:
"""Quyết định có nên retry dựa trên error type"""
retryable_codes = {408, 429, 500, 502, 503, 504}
if hasattr(error, 'response'):
status_code = error.response.status_code
if status_code in retryable_codes:
return True
# Retry network errors (timeout, connection)
if isinstance(error, (ConnectionError, TimeoutError)):
return True
return attempt < self.max_retries
def _wait_with_jitter(self, attempt: int):
"""Exponential backoff với jitter"""
base_delay = 2 ** attempt
jitter = random.uniform(0, 1)
delay = min(base_delay * jitter, 60) # Cap at 60 seconds
print(f"Waiting {delay:.1f}s before retry...")
time.sleep(delay)
def request(
self,
method: str,
url: str,
**kwargs
) -> requests.Response:
"""Execute request với rate limiting, retry và circuit breaker"""
# Check circuit breaker
if self.circuit_open:
if time.time() - self.circuit_open_time > self.circuit_reset_timeout:
print("Circuit breaker: resetting")
self.circuit_open = False
self.failure_count = 0
else:
raise Exception("Circuit breaker is OPEN. Too many failures.")
# Acquire rate limit token
if not self.rate_limiter.acquire(timeout=30):
raise Exception("Rate limiter timeout")
attempt = 0
last_error = None
while attempt < self.max_retries:
try:
response = requests.request(
method=method,
url=url,
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=(10, 120),
**kwargs
)
# Check for rate limit response
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"Rate limited. Waiting {retry_after}s...")
time.sleep(retry_after)
attempt += 1
continue
response.raise_for_status()
# Success - reset circuit breaker
self.failure_count = 0
self.request_history.append({"success": True, "timestamp": time.time()})
return response
except Exception as e:
last_error = e
self.request_history.append({
"success": False,
"error": str(e),
"timestamp": time.time()
})
if self._should_retry(e, attempt):
attempt += 1
self._wait_with_jitter(attempt)
else:
break
# Failed after all retries - open circuit breaker
self.failure_count += 1
if self.failure_count >= self.failure_threshold:
self.circuit_open = True
self.circuit_open_time = time.time()
print(f"Circuit breaker OPENED after {self.failure_count} failures")
raise last_error
Sử dụng client an toàn
client = RobustAPIClient(
api_key="YOUR_TARDIS_API_KEY",
requests_per_second=5, # 5 requests/second
max_retries=5
)
Thực hiện request
try:
response = client.request("GET", "https://api.tardis.io/v1/symbols")
data = response.json()
except Exception as e:
print(f"Request failed: {e}")3. Lỗi "ArrowInvalid: Not a Parquet file" hoặc "Invalid: Parquet file size is 0"
# Nguyên nhân: File download không hoàn chỉnh hoặc file bị corrupt
Cách khắc phục:
import pyarrow.parquet as pq
from pathlib import Path
import hashlib
class ParquetValidator:
"""Validate Parquet files trước khi xử lý"""
def __init__(self, min_size_bytes: int = 100):
self.min_size = min_size_bytes
self.validation_results = {}
def validate_file(self, filepath: Path) -> dict:
"""Validate một Parquet file"""
result = {
"path": str(filepath),
"valid": False,
"size": 0,
"row_groups": 0,
"schema": None,
"errors": []
}
# 1. Check file exists và size
if not filepath.exists():
result["errors"].append("File does not exist")
return result
result["size"] = filepath.stat().st_size
if result["size"] < self.min_size:
result["errors"].append(f"File too small: {result['size']} bytes")
return result
# 2. Try read metadata (fast check)
try:
parquet_file = pq.ParquetFile(str(filepath))
result["row_groups"] = parquet_file.metadata.num_row_groups
result["schema"] = str(parquet_file.schema)
result["total_rows"] = parquet_file.metadata.num_rows
# 3. Validate magic bytes
with open(filepath, "rb") as f:
magic = f.read(4)
if magic != b"PAR1":
result["errors"].append(f"Invalid magic bytes: {magic}")
return result
# 4. Quick read test (first row group only)
table = parquet_file.read_row_group(0)
if table.num_rows == 0:
result["errors"].append("No rows in first row group")
result["valid"] = len(result["errors"]) == 0
except Exception as e:
result["errors"].append(f"PyArrow error: {str(e)}")
return result
def validate_directory(self, data_dir: str, pattern: str = "*.parquet") -> dict:
"""Validate tất cả files trong thư mục"""
data_path = Path(data_dir)
files = list(data_path.glob(pattern))
summary = {
"total_files": len(files),
"valid_files": 0,
"invalid_files": 0,
"total_size": 0,
"results": []
}
for f in files:
result = self.validate_file(f)
summary["results"].append(result)
if result["valid"]:
summary["valid_files"] += 1
else:
summary["invalid_files"] += 1
summary["