All checks were successful
Build and Push Docker Image / build-and-push (release) Successful in 1m16s
75 lines
2.0 KiB
Python
75 lines
2.0 KiB
Python
import time
|
|
from datetime import datetime
|
|
import re
|
|
from random import randint
|
|
|
|
|
|
startTime = None
|
|
|
|
def setStartTime():
|
|
global startTime
|
|
startTime = time.time()
|
|
|
|
def getPnL(pair):
|
|
with open("./data/tradingLog.log", "r") as f:
|
|
lines = f.readlines()
|
|
|
|
logEntries = []
|
|
for i in lines:
|
|
# Get timestamp + message tuples
|
|
t = re.match(r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}) - (.+)", i)
|
|
if t:
|
|
timestamp_str, message = t.groups()
|
|
timestamp = datetime.strptime(timestamp_str, "%Y-%m-%d %H:%M:%S,%f")
|
|
logEntries.append((timestamp, message.strip()))
|
|
|
|
strategyStartTime = None
|
|
for timestamp, message in logEntries:
|
|
if message == (f"Starting strategy with {pair}"):
|
|
strategyStartTime = timestamp
|
|
|
|
if not strategyStartTime:
|
|
print(f"No 'Starting strategy' found for pair {pair}")
|
|
return 0.0
|
|
|
|
totalPnL = 0.0
|
|
pattern = re.compile(
|
|
rf"(Long|Short) order on {re.escape(pair)} level .*? filled with P&L (-?\d+\.?\d*) and qty .*"
|
|
)
|
|
|
|
for timestamp, message in logEntries:
|
|
if timestamp >= strategyStartTime:
|
|
match = pattern.search(message)
|
|
if match:
|
|
t = float(match.group(2))
|
|
totalPnL += t
|
|
|
|
return totalPnL
|
|
|
|
def getLevels(amount, highPrice, lowPrice, roundDecimals):
|
|
# Returns array of prices from low to high for each level
|
|
levels = []
|
|
delta = (highPrice - lowPrice)/amount
|
|
|
|
levelPrice = lowPrice
|
|
for i in range(amount - 1):
|
|
levels.append(levelPrice)
|
|
levelPrice = round(levelPrice + delta, roundDecimals)
|
|
levels.append(highPrice)
|
|
return levels
|
|
|
|
def floor(value, decimals):
|
|
# Rounds float to the lower side with the decimals given
|
|
factor = 1/(10**decimals)
|
|
return (value//factor)*factor
|
|
|
|
def countDecimals(value):
|
|
# Counts decimals in a float
|
|
decimals = len(str(value).split('.')[-1])
|
|
return decimals
|
|
|
|
|
|
def getArbus():
|
|
# Returnes n arbus (n is random 1 byte positive number)
|
|
return ("arbus"*randint(0, 255))
|