This commit is contained in:
David Brazda
2023-08-30 21:19:54 +02:00
parent 8a8a1e19be
commit 28e44b0177
2 changed files with 94 additions and 12 deletions

View File

@@ -140,6 +140,8 @@ def next(data, state: StrategyState):
# v tuplu (nazevind,direktiva,hodnota) # v tuplu (nazevind,direktiva,hodnota)
# do OR jsou dane i bez prefixu # do OR jsou dane i bez prefixu
# {'AND': [('nazev indikatoru', 'nazev direktivy', 'hodnotadirektivy')], 'OR': []} # {'AND': [('nazev indikatoru', 'nazev direktivy', 'hodnotadirektivy')], 'OR': []}
#POZOR TOTO JE STARY FORMAT - podminky jsou uvnitr sekce indikatoru
#v INITU uz mame novy format uvnitr sekce signal v podsekci conditions
def get_work_dict_with_directive(starts_with: str): def get_work_dict_with_directive(starts_with: str):
reslist = dict(AND=[], OR=[]) reslist = dict(AND=[], OR=[])
@@ -670,10 +672,17 @@ def next(data, state: StrategyState):
return tick return tick
def get_default_sl_value(direction: TradeDirection): def get_default_sl_value(direction: TradeDirection):
if direction == TradeDirection.LONG: if direction == TradeDirection.LONG:
smer = "long" smer = "long"
else: else:
smer = "short" smer = "short"
#TODO zda signal, ktery activeTrade vygeneroval, nema vlastni nastaveni + fallback na general
options = safe_get(state.vars, 'exit_conditions', None) options = safe_get(state.vars, 'exit_conditions', None)
if options is None: if options is None:
state.ilog(e="No options for exit conditions in stratvars. Fallback.") state.ilog(e="No options for exit conditions in stratvars. Fallback.")
@@ -683,24 +692,29 @@ def next(data, state: StrategyState):
return val return val
def get_profit_target_price(): def get_profit_target_price():
#TODO zda signal, ktery activeTrade vygeneroval, nema vlastni nastaveni + fallback na general
def_profit = safe_get(state.vars, "def_profit",state.vars.profit) def_profit = safe_get(state.vars, "def_profit",state.vars.profit)
cena = float(state.avgp) cena = float(state.avgp)
return price2dec(cena+normalize_tick(float(state.vars.profit)),3) if int(state.positions) > 0 else price2dec(cena-normalize_tick(float(state.vars.profit)),3) return price2dec(cena+normalize_tick(float(state.vars.profit)),3) if int(state.positions) > 0 else price2dec(cena-normalize_tick(float(state.vars.profit)),3)
def get_max_profit_price(): def get_max_profit_price():
#TODO zda signal, ktery activeTrade vygeneroval, nema vlastni nastaveni + fallback na general
max_profit = float(safe_get(state.vars, "max_profit",0.03)) max_profit = float(safe_get(state.vars, "max_profit",0.03))
cena = float(state.avgp) cena = float(state.avgp)
return price2dec(cena+normalize_tick(max_profit),3) if int(state.positions) > 0 else price2dec(cena-normalize_tick(max_profit),3) return price2dec(cena+normalize_tick(max_profit),3) if int(state.positions) > 0 else price2dec(cena-normalize_tick(max_profit),3)
#TBD pripadne opet dat parsovani pole do INITu #TBD pripadne opet dat parsovani pole do INITu
#TODO nejspis u exitu neporovnavat s MA i kdyz indikator ma, ale s online hodnotou ??
def exit_conditions_met(direction: TradeDirection): def exit_conditions_met(direction: TradeDirection):
if direction == TradeDirection.LONG: if direction == TradeDirection.LONG:
smer = "long" smer = "long"
else: else:
smer = "short" smer = "short"
#TODO zda signal, ktery activeTrade vygeneroval, nema vlastni EXIT nastaveni + fallback na general
options = safe_get(state.vars, 'exit_conditions', None) options = safe_get(state.vars, 'exit_conditions', None)
if options is None: if options is None:
state.ilog(e="No options for exit conditions in stratvars") state.ilog(e="No options for exit conditions in stratvars")
@@ -720,6 +734,8 @@ def next(data, state: StrategyState):
state.ilog(e=f"EXIT_CONDITION for{smer} DISABLED by {conditions_met}", **conditions_met) state.ilog(e=f"EXIT_CONDITION for{smer} DISABLED by {conditions_met}", **conditions_met)
return False return False
#EXIT zatim po staru v indikatorech pro vsechny
work_dict_exit_if = get_work_dict_with_directive(starts_with="exit_"+smer+"_if") work_dict_exit_if = get_work_dict_with_directive(starts_with="exit_"+smer+"_if")
state.ilog(e=f"EXIT CONDITION for {smer} work_dict", message=work_dict_exit_if) state.ilog(e=f"EXIT CONDITION for {smer} work_dict", message=work_dict_exit_if)
@@ -846,6 +862,9 @@ def next(data, state: StrategyState):
state.ilog(e="Eval CLOSE", price=curr_price, pos=state.positions, avgp=state.avgp, pending=state.vars.pending, activeTrade=str(state.vars.activeTrade)) state.ilog(e="Eval CLOSE", price=curr_price, pos=state.positions, avgp=state.avgp, pending=state.vars.pending, activeTrade=str(state.vars.activeTrade))
if int(state.positions) != 0 and float(state.avgp)>0 and state.vars.pending is None: if int(state.positions) != 0 and float(state.avgp)>0 and state.vars.pending is None:
#podivam se zda dana
#pevny target - presunout toto do INIT a pak jen pristupovat #pevny target - presunout toto do INIT a pak jen pristupovat
goal_price = get_profit_target_price() goal_price = get_profit_target_price()
max_price = get_max_profit_price() max_price = get_max_profit_price()
@@ -1014,6 +1033,11 @@ def next(data, state: StrategyState):
work_dict_dont_do = state.vars.work_dict_dont_do[signalname+"_"+ smer] work_dict_dont_do = state.vars.work_dict_dont_do[signalname+"_"+ smer]
state.ilog(e=f"{smer} PRECOND DONT{smer} work_dict for {signalname}", message=work_dict_dont_do) state.ilog(e=f"{smer} PRECOND DONT{smer} work_dict for {signalname}", message=work_dict_dont_do)
#TEMP docasne
state.ilog(e=f"{smer} ALT NA SROVNANI DONT{smer} work_dict for {signalname}", message=state.vars.work_dict_dont_do_new)
#u techto ma smysl pouze OR #u techto ma smysl pouze OR
precond = create_conditions_from_directives(work_dict_dont_do, "OR") precond = create_conditions_from_directives(work_dict_dont_do, "OR")
result, conditions_met = eval_cond_dict(precond) result, conditions_met = eval_cond_dict(precond)
@@ -1051,6 +1075,10 @@ def next(data, state: StrategyState):
state.ilog(e=f"{smer} SIGNAL work_dict {signalname}", message=work_dict_signal_if) state.ilog(e=f"{smer} SIGNAL work_dict {signalname}", message=work_dict_signal_if)
#TEMP DOCASNE
state.ilog(e=f"{smer} ALT NA SROVNANI SIGNAL work_dict {signalname}", message=state.vars.work_dict_signal_if_new)
state.ilog(e=f"ACTIVATION work_dict {signalname}", message=state.vars.work_dict_signal_activate_if)
buy_or_cond = create_conditions_from_directives(work_dict_signal_if, "OR") buy_or_cond = create_conditions_from_directives(work_dict_signal_if, "OR")
result, conditions_met = eval_cond_dict(buy_or_cond) result, conditions_met = eval_cond_dict(buy_or_cond)
state.ilog(e=f"{smer} SIGNAL =OR= {result}", **conditions_met) state.ilog(e=f"{smer} SIGNAL =OR= {result}", **conditions_met)
@@ -1102,6 +1130,7 @@ def next(data, state: StrategyState):
id=uuid4(), id=uuid4(),
last_update=datetime.fromtimestamp(state.time).astimezone(zoneNY), last_update=datetime.fromtimestamp(state.time).astimezone(zoneNY),
status=TradeStatus.READY, status=TradeStatus.READY,
generated_by=name,
direction=TradeDirection.LONG, direction=TradeDirection.LONG,
entry_price=None, entry_price=None,
stoploss_value = None)) stoploss_value = None))
@@ -1110,6 +1139,7 @@ def next(data, state: StrategyState):
id=uuid4(), id=uuid4(),
last_update=datetime.fromtimestamp(state.time).astimezone(zoneNY), last_update=datetime.fromtimestamp(state.time).astimezone(zoneNY),
status=TradeStatus.READY, status=TradeStatus.READY,
generated_by=name,
direction=TradeDirection.SHORT, direction=TradeDirection.SHORT,
entry_price=None, entry_price=None,
stoploss_value = None)) stoploss_value = None))
@@ -1179,10 +1209,12 @@ def init(state: StrategyState):
#place to declare new vars #place to declare new vars
print("INIT v main",state.name) print("INIT v main",state.name)
#pomocna funkce pro inicializaci #pomocna funkce pro inicializaci - stary format uvnitr indikatoru (napr. trendfollow_long_if, AND.trendfollow_long_if)
def get_work_dict_with_directive(starts_with: str): def get_work_dict_with_directive(starts_with: str):
reslist = dict(AND=[], OR=[]) reslist = dict(AND=[], OR=[])
#vraci v kazdem klici truple (indname, volba, hodnota)
#vracime podle puvodniho formatu direktiv uvnitr indikatoru
for indname, indsettings in state.vars.indicators.items(): for indname, indsettings in state.vars.indicators.items():
for option,value in indsettings.items(): for option,value in indsettings.items():
if option.startswith(starts_with): if option.startswith(starts_with):
@@ -1199,6 +1231,41 @@ def init(state: StrategyState):
reslist["OR"].append((indname, key, val)) reslist["OR"].append((indname, key, val))
return reslist return reslist
#pomocna funkce pro vytvoreni podminkoveho directory z direktiv v novem formatu
# direktiva v CONDITIONS sekci u daneho SIGNALU
#
# stratvars.signals.trendfollow.conditions
# slope30.short_if_below = 0.3
# slope20.AND.lonf_if_above = 2
# slope30.AND.ACTION_if_above
# ACTION: long_, short_, exit_ (action)
# AND: optional operator AND/OR
def get_work_dict_from_signal_section(action: str, signal_name: str):
reslist = dict(AND=[], OR=[])
try:
for indname, condition in state.vars.signals[signal_name]["conditions"].items():
#prvnim je vzdy indikator na ktery se direktiva odkazuje, tzn. projedeme vsechny tyto indikatory
for directive, value in condition.items():
if directive.startswith(action):
reslist["OR"].append((indname, directive, value))
if directive == "AND":
#vsechny buy direktivy, ktere jsou pod AND
for key, val in value.items():
if key.startswith(action):
reslist["AND"].append((indname, key, val))
if directive == "OR" :
#vsechny buy direktivy, ktere jsou pod OR
for key, val in value.items():
if key.startswith(action):
reslist["OR"].append((indname, key, val))
except KeyError:
pass
#vysledek: v kazdem klici truple (indname, volba, hodnota)
return reslist
def initialize_dynamic_indicators(): def initialize_dynamic_indicators():
#pro vsechny indikatory, ktere maji ve svych stratvars TYPE inicializujeme #pro vsechny indikatory, ktere maji ve svych stratvars TYPE inicializujeme
for indname, indsettings in state.vars.indicators.items(): for indname, indsettings in state.vars.indicators.items():
@@ -1215,16 +1282,29 @@ def init(state: StrategyState):
state.statinds[indname] = dict(minimum_slope=safe_get(indsettings, 'minimum_slope', -1), maximum_slope=safe_get(indsettings, 'maximum_slope', 1)) state.statinds[indname] = dict(minimum_slope=safe_get(indsettings, 'minimum_slope', -1), maximum_slope=safe_get(indsettings, 'maximum_slope', 1))
def intialize_work_dict(): def intialize_work_dict():
#inciializace pro akce: short, long, dont_short, dont_long, activate
state.vars
state.vars.work_dict_dont_do = {} state.vars.work_dict_dont_do = {}
state.vars.work_dict_signal_if = {} state.vars.work_dict_signal_if = {}
state.vars.work_dict_dont_do_new = {}
state.vars.work_dict_signal_if_new = {}
state.vars.work_dict_signal_activate_if = {}
#pro kazdy signal si vytvorime promenou, ktera bude obsahovat direktivy podminek pro akce activation, dont doshort/long and short/long
for signalname, signalsettings in state.vars.signals.items(): for signalname, signalsettings in state.vars.signals.items():
smer = TradeDirection.LONG #TEMP: direktiva pro aktivaci zatim wip
state.vars.work_dict_dont_do[signalname+"_"+ smer] = get_work_dict_with_directive(starts_with=signalname+"_dont_"+ smer +"_if") state.vars.work_dict_signal_activate_if[signalname] = get_work_dict_from_signal_section(action="activate_if", signal_name=signalname)
state.vars.work_dict_signal_if[signalname+"_"+ smer] = get_work_dict_with_directive(starts_with=signalname+"_"+smer+"_if")
smer = TradeDirection.SHORT for smer in TradeDirection:
state.vars.work_dict_dont_do[signalname+"_"+ smer] = get_work_dict_with_directive(starts_with=signalname+"_dont_"+ smer +"_if") state.vars.work_dict_dont_do[signalname+"_"+ smer] = get_work_dict_with_directive(starts_with=signalname+"_dont_"+ smer +"_if")
state.vars.work_dict_signal_if[signalname+"_"+ smer] = get_work_dict_with_directive(starts_with=signalname+"_"+smer+"_if") state.vars.work_dict_signal_if[signalname+"_"+ smer] = get_work_dict_with_directive(starts_with=signalname+"_"+smer+"_if")
#TEMP:zatim nove direktivy pro porovnanin
state.vars.work_dict_dont_do_new[signalname+"_"+ smer] = get_work_dict_from_signal_section(action="dont_" + smer +"_if", signal_name=signalname)
state.vars.work_dict_signal_if_new[signalname+"_"+ smer] = get_work_dict_from_signal_section(action=smer +"_if", signal_name=signalname)
#init klice v extData pro ulozeni historie SL #init klice v extData pro ulozeni historie SL
state.extData["sl_history"] = [] state.extData["sl_history"] = []

View File

@@ -589,9 +589,11 @@ function chart_archived_run(archRecord, data, oneMinuteBars) {
chart.removeSeries(markersLine) chart.removeSeries(markersLine)
} }
if (slLine) {
slLine.forEach((series, index, array) => { slLine.forEach((series, index, array) => {
chart.removeSeries(series) chart.removeSeries(series)
}) })
}
// if (slLine) { // if (slLine) {
// chart.removeSeries(slLine) // chart.removeSeries(slLine)
// } // }