Python JSON 序列化字典<String,Object>

Question

我有一个带有键字符串 (2019-10-28 13:21) 和对象值 (DataPoint) 的字典

import requests
import json
import time

symbol = "AAPL"

intraday_url = "https://www.alphavantage.co/query?function=TIME_SERIES_INTRADAY&symbol="+symbol+"&interval=1min&outputsize=full&apikey="+api_key
sma_url = "https://www.alphavantage.co/query?function=SMA&symbol="+symbol+"&interval=1min&time_period=180&series_type=open&apikey="+api_key
ema_url = "https://www.alphavantage.co/query?function=EMA&symbol="+symbol+"&interval=1min&time_period=15&series_type=open&apikey="+api_key
vwap_url = "https://www.alphavantage.co/query?function=VWAP&symbol="+symbol+"&interval=1min&apikey="+api_key
macd_url = "https://www.alphavantage.co/query?function=MACD&symbol="+symbol+"&interval=1min&series_type=open&apikey="+api_key
rsi_url = "https://www.alphavantage.co/query?function=RSI&symbol="+symbol+"&interval=1min&time_period=100&series_type=open&apikey="+api_key
adx_url = "https://www.alphavantage.co/query?function=ADX&symbol="+symbol+"&interval=1min&time_period=100&apikey="+api_key

class DataPoint:
    def __init__(self, time):
        # 2019-10-31 15:49:00 (original)
        # 2019-10-31 15:49 (formatted)
        formatted_time = time[0:len(time)-3]
        self.time = formatted_time
        self.open = None
        self.high = None
        self.low = None
        self.close = None
        self.volume = None
        self.sma = None
        self.ema = None
        self.vwap = None
        self.macd = None
        self.rsi = None
        self.adx = None

    def addIntraday(self,open,high,low,close,volume):
        self.open = open
        self.high = high
        self.low = low
        self.close = close
        self.volume = volume

    def addTechnical(self,technical,value):
        if technical == "SMA":
            self.sma = value
        elif technical == "EMA":
            self.ema = value
        elif technical == "VWAP":
            self.vwap = value
        elif technical == "MACD":
            self.macd = value
        elif technical == "RSI":
            self.rsi = value
        elif technical == "ADX":
            self.adx = value

def getIntraday(dictionary):
    url = intraday_url
    response = requests.get(url)
    json = response.json()
    intraday = json.get("Time Series (1min)")
    keys = intraday.keys()
    for key in keys:
        ts = intraday.get(key)
        dp = DataPoint(key)
        open = ts.get("1. open")
        high = ts.get("2. high")
        low = ts.get("3. low")
        close = ts.get("4. close")
        volume = ts.get("5. volume")
        dp.addIntraday(open,high,low,close,volume)
        dictionary[dp.time] = dp

def getTechnicals(dictionary):
    urls = [sma_url, ema_url, vwap_url, macd_url, rsi_url, adx_url]
    technicals = ["SMA","EMA","VWAP","MACD","RSI","ADX"]
    i = 0
    while (i < len(urls)):
        response = requests.get(urls[i])
        json = response.json()
        tech = json.get("Technical Analysis: " + technicals[i])
        if (tech == None):
            print("Empty response, retrying in 10 seconds...")
            time.sleep(10)
        else:
            print("Getting Technical Indicator: " + technicals[i])
            keys = tech.keys()
            for key in keys:
                t = tech.get(key)
                v = t.get(technicals[i])
                if (dictionary.get(key) != None):
                    dictionary.get(key).addTechnical(technicals[i], v)
            i += 1

def writeDictionaryToFile(dictionary):
    filename = "datapoints.json"
    fp = open(filename, "a")
    json_dictionary = json.dumps(dictionary)
    fp.write(json_dictionary)
    print("Wrote results to file: " + filename)

dictionary = {}

getIntraday(dictionary)

getTechnicals(dictionary)

writeDictionaryToFile(dictionary)

这是错误：

Traceback (most recent call last):
  File "/Users/Jason/Dev/Python/neural-network-example/alphavantage.py", line 124, in <module>
    writeDictionaryToFile(dictionary)
  File "/Users/Jason/Dev/Python/neural-network-example/alphavantage.py", line 113, in writeDictionaryToFile
    json_dictionary = json.dumps(dictionary)
  File "/usr/local/Cellar/python/3.7.4_1/Frameworks/Python.framework/Versions/3.7/lib/python3.7/json/__init__.py", line 231, in dumps
    return _default_encoder.encode(obj)
  File "/usr/local/Cellar/python/3.7.4_1/Frameworks/Python.framework/Versions/3.7/lib/python3.7/json/encoder.py", line 199, in encode
    chunks = self.iterencode(o, _one_shot=True)
  File "/usr/local/Cellar/python/3.7.4_1/Frameworks/Python.framework/Versions/3.7/lib/python3.7/json/encoder.py", line 257, in iterencode
    return _iterencode(o, 0)
  File "/usr/local/Cellar/python/3.7.4_1/Frameworks/Python.framework/Versions/3.7/lib/python3.7/json/encoder.py", line 179, in default
    raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type DataPoint is not JSON serializable

据我了解，我可以在常见的 python 数据类型、字符串、int、数组字典等上使用 json.dumps()。但我不能在我创建的自定义对象上使用它。我已经完成了研究，并且从我的研究中，我想出了使用 myCustomObject.dict 来使对象可序列化。当我尝试序列化整个字典时如何使用它？

我是 Python 新手，我只是想不通。非常感谢任何帮助。

Answer 1

可以使用自定义 JSON 序列化程序来实现此目的，但这可能对您的任务来说太过分了。一个更简单的解决方案是通过字典为您的类提供几个方法来转换为 JSON 并返回。

你的类有很多字段，所以我将举一个玩具类的更简单的例子，你应该能够适应你的目的：

import json

class Example:
    def __init__(self, x, y):
        self.x = x
        self.y = y

    def to_json(self):
        return json.dumps({
            'x': self.x,
            'y': self.y
        })

    @classmethod
    def from_json(cls, s):
        d = json.loads(s)
        return cls(d['x'], d['y'])

用法：

>>> ex = Example(1, 2)
>>> s = ex.to_json()
>>> s
'{"y": 2, "x": 1}'
>>> ex2 = Example.from_json(s)
>>> ex2.x
1
>>> ex2.y
2

Answer 2

json 是一种非常便携的格式，但它也受到限制。只有以下的东西可以用Json序列化：

dicts - {} (all keys must be strings)
lists - []
strings - "string"
integers - 0, 1, 2, ...
True
False
None

所以你必须将你的对象转换为这些东西的某种组合，并有代码将其转换回来。

如果你只打算使用 python，你可能对pickle 感兴趣，它可以序列化任意 python 对象，只要它可以导入定义它们的模块。请注意，从未知来源解包泡菜可能会导致远程代码执行。