#155 + presun row_to from db.py to transform.py

This commit is contained in:
David Brazda
2024-03-06 12:55:36 +01:00
parent b4ac17585b
commit ce6dc58764
3 changed files with 96 additions and 94 deletions

View File

@ -2,12 +2,6 @@ import sqlite3
import queue import queue
import threading import threading
import time import time
from v2realbot.common.model import RunArchive, RunArchiveView, RunManagerRecord
from datetime import datetime
import orjson
from v2realbot.utils.utils import json_serial, send_to_telegram, zoneNY
import v2realbot.controller.services as cs
from uuid import UUID
from v2realbot.config import DATA_DIR from v2realbot.config import DATA_DIR
sqlite_db_file = DATA_DIR + "/v2trading.db" sqlite_db_file = DATA_DIR + "/v2trading.db"
@ -64,86 +58,3 @@ pool = ConnectionPool(10)
#for one shared connection (used for writes only in WAL mode) #for one shared connection (used for writes only in WAL mode)
insert_conn = sqlite3.connect(sqlite_db_file, check_same_thread=False) insert_conn = sqlite3.connect(sqlite_db_file, check_same_thread=False)
insert_queue = queue.Queue() insert_queue = queue.Queue()
#prevede dict radku zpatky na objekt vcetme retypizace
def row_to_runmanager(row: dict) -> RunManagerRecord:
is_running = cs.is_runner_running(row['runner_id']) if row['runner_id'] else False
res = RunManagerRecord(
moddus=row['moddus'],
id=row['id'],
strat_id=row['strat_id'],
symbol=row['symbol'],
mode=row['mode'],
account=row['account'],
note=row['note'],
ilog_save=bool(row['ilog_save']),
bt_from=datetime.fromisoformat(row['bt_from']) if row['bt_from'] else None,
bt_to=datetime.fromisoformat(row['bt_to']) if row['bt_to'] else None,
weekdays_filter=[int(x) for x in row['weekdays_filter'].split(',')] if row['weekdays_filter'] else [],
batch_id=row['batch_id'],
testlist_id=row['testlist_id'],
start_time=row['start_time'],
stop_time=row['stop_time'],
status=row['status'],
#last_started=zoneNY.localize(datetime.fromisoformat(row['last_started'])) if row['last_started'] else None,
last_processed=datetime.fromisoformat(row['last_processed']) if row['last_processed'] else None,
history=row['history'],
valid_from=datetime.fromisoformat(row['valid_from']) if row['valid_from'] else None,
valid_to=datetime.fromisoformat(row['valid_to']) if row['valid_to'] else None,
runner_id = row['runner_id'] if row['runner_id'] and is_running else None, #runner_id is only present if it is running
strat_running = is_running) #cant believe this when called from separate process as not current
return res
#prevede dict radku zpatky na objekt vcetme retypizace
def row_to_runarchiveview(row: dict) -> RunArchiveView:
a = RunArchiveView(
id=row['runner_id'],
strat_id=row['strat_id'],
batch_id=row['batch_id'],
symbol=row['symbol'],
name=row['name'],
note=row['note'],
started=datetime.fromisoformat(row['started']) if row['started'] else None,
stopped=datetime.fromisoformat(row['stopped']) if row['stopped'] else None,
mode=row['mode'],
account=row['account'],
bt_from=datetime.fromisoformat(row['bt_from']) if row['bt_from'] else None,
bt_to=datetime.fromisoformat(row['bt_to']) if row['bt_to'] else None,
ilog_save=bool(row['ilog_save']),
profit=float(row['profit']),
trade_count=int(row['trade_count']),
end_positions=int(row['end_positions']),
end_positions_avgp=float(row['end_positions_avgp']),
metrics=orjson.loads(row['metrics']) if row['metrics'] else None,
batch_profit=int(row['batch_profit']) if row['batch_profit'] and row['batch_id'] else 0,
batch_count=int(row['batch_count']) if row['batch_count'] and row['batch_id'] else 0,
)
return a
#prevede dict radku zpatky na objekt vcetme retypizace
def row_to_runarchive(row: dict) -> RunArchive:
return RunArchive(
id=row['runner_id'],
strat_id=row['strat_id'],
batch_id=row['batch_id'],
symbol=row['symbol'],
name=row['name'],
note=row['note'],
started=datetime.fromisoformat(row['started']) if row['started'] else None,
stopped=datetime.fromisoformat(row['stopped']) if row['stopped'] else None,
mode=row['mode'],
account=row['account'],
bt_from=datetime.fromisoformat(row['bt_from']) if row['bt_from'] else None,
bt_to=datetime.fromisoformat(row['bt_to']) if row['bt_to'] else None,
strat_json=orjson.loads(row['strat_json']),
settings=orjson.loads(row['settings']),
ilog_save=bool(row['ilog_save']),
profit=float(row['profit']),
trade_count=int(row['trade_count']),
end_positions=int(row['end_positions']),
end_positions_avgp=float(row['end_positions_avgp']),
metrics=orjson.loads(row['metrics']),
stratvars_toml=row['stratvars_toml']
)

View File

@ -0,0 +1,87 @@
from v2realbot.common.model import RunArchive, RunArchiveView, RunManagerRecord
from datetime import datetime
import orjson
import v2realbot.controller.services as cs
#prevede dict radku zpatky na objekt vcetme retypizace
def row_to_runmanager(row: dict) -> RunManagerRecord:
is_running = cs.is_runner_running(row['runner_id']) if row['runner_id'] else False
res = RunManagerRecord(
moddus=row['moddus'],
id=row['id'],
strat_id=row['strat_id'],
symbol=row['symbol'],
mode=row['mode'],
account=row['account'],
note=row['note'],
ilog_save=bool(row['ilog_save']),
bt_from=datetime.fromisoformat(row['bt_from']) if row['bt_from'] else None,
bt_to=datetime.fromisoformat(row['bt_to']) if row['bt_to'] else None,
weekdays_filter=[int(x) for x in row['weekdays_filter'].split(',')] if row['weekdays_filter'] else [],
batch_id=row['batch_id'],
testlist_id=row['testlist_id'],
start_time=row['start_time'],
stop_time=row['stop_time'],
status=row['status'],
#last_started=zoneNY.localize(datetime.fromisoformat(row['last_started'])) if row['last_started'] else None,
last_processed=datetime.fromisoformat(row['last_processed']) if row['last_processed'] else None,
history=row['history'],
valid_from=datetime.fromisoformat(row['valid_from']) if row['valid_from'] else None,
valid_to=datetime.fromisoformat(row['valid_to']) if row['valid_to'] else None,
runner_id = row['runner_id'] if row['runner_id'] and is_running else None, #runner_id is only present if it is running
strat_running = is_running) #cant believe this when called from separate process as not current
return res
#prevede dict radku zpatky na objekt vcetme retypizace
def row_to_runarchiveview(row: dict) -> RunArchiveView:
a = RunArchiveView(
id=row['runner_id'],
strat_id=row['strat_id'],
batch_id=row['batch_id'],
symbol=row['symbol'],
name=row['name'],
note=row['note'],
started=datetime.fromisoformat(row['started']) if row['started'] else None,
stopped=datetime.fromisoformat(row['stopped']) if row['stopped'] else None,
mode=row['mode'],
account=row['account'],
bt_from=datetime.fromisoformat(row['bt_from']) if row['bt_from'] else None,
bt_to=datetime.fromisoformat(row['bt_to']) if row['bt_to'] else None,
ilog_save=bool(row['ilog_save']),
profit=float(row['profit']),
trade_count=int(row['trade_count']),
end_positions=int(row['end_positions']),
end_positions_avgp=float(row['end_positions_avgp']),
metrics=orjson.loads(row['metrics']) if row['metrics'] else None,
batch_profit=int(row['batch_profit']) if row['batch_profit'] and row['batch_id'] else 0,
batch_count=int(row['batch_count']) if row['batch_count'] and row['batch_id'] else 0,
)
return a
#prevede dict radku zpatky na objekt vcetme retypizace
def row_to_runarchive(row: dict) -> RunArchive:
return RunArchive(
id=row['runner_id'],
strat_id=row['strat_id'],
batch_id=row['batch_id'],
symbol=row['symbol'],
name=row['name'],
note=row['note'],
started=datetime.fromisoformat(row['started']) if row['started'] else None,
stopped=datetime.fromisoformat(row['stopped']) if row['stopped'] else None,
mode=row['mode'],
account=row['account'],
bt_from=datetime.fromisoformat(row['bt_from']) if row['bt_from'] else None,
bt_to=datetime.fromisoformat(row['bt_to']) if row['bt_to'] else None,
strat_json=orjson.loads(row['strat_json']),
settings=orjson.loads(row['settings']),
ilog_save=bool(row['ilog_save']),
profit=float(row['profit']),
trade_count=int(row['trade_count']),
end_positions=int(row['end_positions']),
end_positions_avgp=float(row['end_positions_avgp']),
metrics=orjson.loads(row['metrics']),
stratvars_toml=row['stratvars_toml']
)

View File

@ -23,6 +23,7 @@ from traceback import format_exc
from datetime import timedelta, time from datetime import timedelta, time
from threading import Lock from threading import Lock
import v2realbot.common.db as db import v2realbot.common.db as db
import v2realbot.common.transform as tr
from sqlite3 import OperationalError, Row from sqlite3 import OperationalError, Row
import v2realbot.strategyblocks.indicators.custom as ci import v2realbot.strategyblocks.indicators.custom as ci
from v2realbot.strategyblocks.inits.init_indicators import initialize_dynamic_indicators from v2realbot.strategyblocks.inits.init_indicators import initialize_dynamic_indicators
@ -114,7 +115,7 @@ def fetch_all_run_manager_records() -> list[RunManagerRecord]:
#Transform row to object #Transform row to object
for row in rows: for row in rows:
#add transformed object into result list #add transformed object into result list
results.append(db.row_to_runmanager(row)) results.append(tr.row_to_runmanager(row))
return 0, results return 0, results
finally: finally:
@ -133,7 +134,7 @@ def fetch_run_manager_record_by_id(strategy_id) -> RunManagerRecord:
if row is None: if row is None:
return -2, "not found" return -2, "not found"
else: else:
return 0, db.row_to_runmanager(row) return 0, tr.row_to_runmanager(row)
except Exception as e: except Exception as e:
print("ERROR while fetching all records:", str(e) + format_exc()) print("ERROR while fetching all records:", str(e) + format_exc())
@ -153,6 +154,9 @@ def add_run_manager_record(new_record: RunManagerRecord):
if new_record.stop_time is None: if new_record.stop_time is None:
return -2, f"Invalid stop_time format {new_record.stop_time}" return -2, f"Invalid stop_time format {new_record.stop_time}"
if new_record.batch_id is None:
new_record.batch_id = str(uuid4())[:8]
conn = db.pool.get_connection() conn = db.pool.get_connection()
try: try:
@ -355,7 +359,7 @@ def fetch_scheduled_candidates_for_start_and_stop(market_datetime_now, market) -
start_candidates = [] start_candidates = []
stop_candidates = [] stop_candidates = []
for row in rows: for row in rows:
run_manager_record = db.row_to_runmanager(row) run_manager_record = tr.row_to_runmanager(row)
if row['is_start_time']: if row['is_start_time']:
start_candidates.append(run_manager_record) start_candidates.append(run_manager_record)
if row['is_stop_time']: if row['is_stop_time']:
@ -426,7 +430,7 @@ def fetch_startstop_scheduled_candidates(market_datetime_now, time_check, market
""" """
cursor.execute(query, (market_datetime_now_str, market_datetime_now_str, current_time_str)) cursor.execute(query, (market_datetime_now_str, market_datetime_now_str, current_time_str))
rows = cursor.fetchall() rows = cursor.fetchall()
results = [db.row_to_runmanager(row) for row in rows] results = [tr.row_to_runmanager(row) for row in rows]
return 0, results return 0, results
except Exception as e: except Exception as e: