File size: 2,680 Bytes
7a8d58d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
"""
NanEncoder - Custom JSON serializer for vectorbt output.
Handles NaN, Inf, datetime, numpy types, and pandas objects.
"""

import json
import math
import numpy as np
import pandas as pd
from datetime import datetime, date


class NanEncoder(json.JSONEncoder):
    """Custom JSON encoder that handles vectorbt/pandas/numpy edge cases."""

    def default(self, obj):
        # numpy types
        if isinstance(obj, (np.integer,)):
            return int(obj)
        if isinstance(obj, (np.floating,)):
            if np.isnan(obj) or np.isinf(obj):
                return None
            return float(obj)
        if isinstance(obj, np.ndarray):
            return obj.tolist()
        if isinstance(obj, (np.bool_,)):
            return bool(obj)

        # pandas types
        if isinstance(obj, pd.Timestamp):
            return obj.isoformat()
        if isinstance(obj, pd.Series):
            return obj.to_dict()
        if isinstance(obj, pd.DataFrame):
            return obj.to_dict(orient="records")
        if isinstance(obj, pd.Timedelta):
            return str(obj)

        # datetime
        if isinstance(obj, (datetime, date)):
            return obj.isoformat()

        # timedelta
        from datetime import timedelta
        if isinstance(obj, timedelta):
            return str(obj)

        return super().default(obj)


def safe_json_response(data: dict) -> dict:
    """
    Convert a dict with numpy/pandas values to JSON-safe dict.
    Replaces NaN/Inf with None, converts timestamps to ISO strings.
    """
    def _clean(value):
        if isinstance(value, dict):
            return {k: _clean(v) for k, v in value.items()}
        if isinstance(value, (list, tuple)):
            return [_clean(v) for v in value]
        if isinstance(value, (np.floating, float)):
            if math.isnan(value) or math.isinf(value):
                return None
            return float(value)
        if isinstance(value, (np.integer,)):
            return int(value)
        if isinstance(value, (np.bool_,)):
            return bool(value)
        if isinstance(value, np.ndarray):
            return _clean(value.tolist())
        if isinstance(value, pd.Timestamp):
            return value.isoformat()
        if isinstance(value, (datetime, date)):
            return value.isoformat()
        if isinstance(value, pd.Timedelta):
            return str(value)
        if isinstance(value, pd.Series):
            return _clean(value.to_dict())
        if isinstance(value, pd.DataFrame):
            return _clean(value.to_dict(orient="records"))
        if pd.isna(value):
            return None
        return value

    return _clean(data)