133 lines
3.7 KiB
Python
133 lines
3.7 KiB
Python
"""
|
|
Error Recovery Utilities.
|
|
|
|
This module provides utilities for saving partial analysis state when errors occur,
|
|
allowing users to resume or inspect work completed before the error.
|
|
|
|
Functions:
|
|
save_partial_analysis: Save partial state to JSON file
|
|
get_partial_analysis_filename: Generate filename for partial analysis
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
|
|
def save_partial_analysis(state: Dict[str, Any], output_file: str) -> None:
|
|
"""
|
|
Save partial analysis state to a JSON file.
|
|
|
|
Handles non-serializable objects by converting them to strings.
|
|
Creates parent directories if they don't exist.
|
|
|
|
Args:
|
|
state: Dictionary containing partial analysis state
|
|
output_file: Path where to save the JSON file
|
|
|
|
Raises:
|
|
PermissionError: If unable to write to output_file location
|
|
OSError: If unable to create parent directories
|
|
|
|
Example:
|
|
>>> state = {
|
|
... "ticker": "AAPL",
|
|
... "error": "Rate limit exceeded",
|
|
... "analyst_reports": {"market": {...}}
|
|
... }
|
|
>>> save_partial_analysis(state, "./results/partial_AAPL.json")
|
|
"""
|
|
# Create parent directory if it doesn't exist
|
|
output_path = Path(output_file)
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Convert state to JSON-serializable format
|
|
serializable_state = _make_serializable(state)
|
|
|
|
# Write to file
|
|
with open(output_file, 'w', encoding='utf-8') as f:
|
|
json.dump(serializable_state, f, indent=2, ensure_ascii=False)
|
|
|
|
|
|
def get_partial_analysis_filename(
|
|
ticker: str,
|
|
timestamp: Optional[datetime] = None,
|
|
output_dir: Optional[str] = None,
|
|
) -> str:
|
|
"""
|
|
Generate a filename for partial analysis output.
|
|
|
|
Format: partial_analysis_{ticker}_{timestamp}.json
|
|
|
|
Args:
|
|
ticker: Stock ticker symbol
|
|
timestamp: Timestamp for filename (default: now)
|
|
output_dir: Output directory (default: TRADINGAGENTS_RESULTS_DIR or ./results)
|
|
|
|
Returns:
|
|
str: Full path to partial analysis file
|
|
|
|
Example:
|
|
>>> get_partial_analysis_filename("AAPL")
|
|
'./results/partial_analysis_AAPL_20241226_103045.json'
|
|
"""
|
|
if timestamp is None:
|
|
timestamp = datetime.now()
|
|
|
|
if output_dir is None:
|
|
output_dir = os.getenv("TRADINGAGENTS_RESULTS_DIR", "./results")
|
|
|
|
# Format: partial_analysis_{ticker}_{YYYYMMDD_HHMMSS}.json
|
|
timestamp_str = timestamp.strftime("%Y%m%d_%H%M%S")
|
|
filename = f"partial_analysis_{ticker}_{timestamp_str}.json"
|
|
|
|
return str(Path(output_dir) / filename)
|
|
|
|
|
|
def _make_serializable(obj: Any) -> Any:
|
|
"""
|
|
Recursively convert objects to JSON-serializable format.
|
|
|
|
Handles:
|
|
- Dictionaries (recurse on values)
|
|
- Lists/tuples (recurse on items)
|
|
- datetime objects (convert to ISO format)
|
|
- Other objects (convert to string)
|
|
|
|
Args:
|
|
obj: Object to make serializable
|
|
|
|
Returns:
|
|
JSON-serializable version of obj
|
|
"""
|
|
if obj is None:
|
|
return None
|
|
|
|
if isinstance(obj, (str, int, float, bool)):
|
|
return obj
|
|
|
|
if isinstance(obj, dict):
|
|
return {key: _make_serializable(value) for key, value in obj.items()}
|
|
|
|
if isinstance(obj, (list, tuple)):
|
|
return [_make_serializable(item) for item in obj]
|
|
|
|
if isinstance(obj, datetime):
|
|
return obj.isoformat()
|
|
|
|
# For everything else (including Mock objects), convert to string
|
|
try:
|
|
# Try to convert to dict if it has __dict__
|
|
if hasattr(obj, '__dict__'):
|
|
return {
|
|
'_type': obj.__class__.__name__,
|
|
'_str': str(obj),
|
|
}
|
|
except Exception:
|
|
pass
|
|
|
|
# Final fallback: convert to string
|
|
return str(obj)
|