?
Here is the CSV-based version, however I can't figure out why, but sometimes the entire file isn't flushed. This version works much faster.
import backtrader as bt
import re
import os
import os.path
from typing import List
import datetime as dt
import logging
import math
import pytz
import csv
def getTableName(data):
name = getattr(data,'_name',data.__class__.__name__).strip()
# Replace spaces with underscores
name = re.sub('\s+','_',name)
# Remove invalid characters
name = re.sub('[^0-9a-zA-Z_]', '', name)
# Remove leading characters until we find a letter or underscore
name = re.sub('^[^a-zA-Z_]+', '', name)
return name
def getTableColumns(data):
columns = dict(
datetime = dt.datetime
)
for line in data.lines.getlinealiases():
if line == 'datetime':
continue
columns[line] = float
return columns
class PersistToCSV(bt.Observer):
lines = ('ignored',)
params = dict(
debug = False,
rootpath = '/path/to/csv/output/root/'
)
def __init__(self):
bt.Observer.__init__(self)
if not os.path.isdir(self.p.rootpath):
os.makedirs(self.p.rootpath)
self._csvs:List[csv.DictWriter] = []
self._metadata = []
self._objects = self._owner.datas + list(self._owner.observers)
for obj in self._objects:
name = getTableName(obj)
columns = getTableColumns(obj)
path = os.path.join(self.p.rootpath,name+'.csv')
metadata = (name,columns,open(path,'a'),)
exists = os.path.isfile(path)
csvfile = csv.DictWriter(metadata[2],[c for (c,_) in columns.items()])
if not exists:
csvfile.writeheader()
self._metadata.append(metadata)
self._csvs.append(csvfile)
def next(self):
for (obj,meta,csv) in zip(self._objects,self._metadata,self._csvs):
try:
self._persist(obj,meta,csv)
except:
logging.exception("EXCEPTION OCCURRED WHILE WRITING TO CSV")
def _persist(self,data:bt.DataBase,meta,csv:csv.DictWriter):
name,columns = meta[0],meta[1]
row = dict()
foundValue = False
for name,column in columns.items():
column:Column
value = getattr(data.lines,name,None)
if value is not None:
if column == dt.datetime:
value = data.datetime.datetime(0)
# # Convert to UTC
if data.contractdetails is not None:
tz = pytz.timezone(data.contractdetails.m_timeZoneId)
value = tz.normalize(tz.localize(value)).astimezone(pytz.utc)
else:
value = value[0]
if type(value) == float and math.isnan(value):
print("HI")
continue
foundValue = True
row[name] = value
if len(getattr(data,'datetime',[])) <= 1:
data = self._owner.data0
value = data.datetime.datetime(0)
if data.contractdetails is not None:
tz = pytz.timezone(data.contractdetails.m_timeZoneId)
value = tz.normalize(tz.localize(value)).astimezone(pytz.utc)
row['datetime'] = value
if foundValue:
csv.writerow(row)
def stop(self):
for (name,columns,fd) in self._metadata:
fd.flush()
fd.close()
