dwker/v2realbot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

retreat commit

Browse Source
This commit is contained in:
David Brázda
2023-10-09 09:15:52 +02:00
parent a6678f9a4f
commit be93c17848
93 changed files with 16821 additions and 2561 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
v2realbot/strategyblocks/__init__.py Normal file
View File

9
v2realbot/strategyblocks/activetrade/activetrade_hub.py Normal file
View File

@ -0,0 +1,9 @@
from v2realbot.strategyblocks.activetrade.sl.trailsl import trail_SL_management
from v2realbot.strategyblocks.activetrade.close.evaluate_close import eval_close_position
def manage_active_trade(state, data):
trade = state.vars.activeTrade
if trade is None:
return -1
trail_SL_management(state, data)
eval_close_position(state, data)

45
v2realbot/strategyblocks/activetrade/close/close_position.py Normal file
View File

@ -0,0 +1,45 @@
from v2realbot.strategy.base import StrategyState
from v2realbot.strategy.StrategyOrderLimitVykladaciNormalizedMYSELL import StrategyOrderLimitVykladaciNormalizedMYSELL
from v2realbot.enums.enums import RecordType, StartBarAlign, Mode, Account, Followup
from v2realbot.common.PrescribedTradeModel import Trade, TradeDirection, TradeStatus
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.utils.directive_utils import get_conditions_from_configuration
from v2realbot.ml.mlutils import load_model
from v2realbot.common.model import SLHistory
from v2realbot.config import KW
from uuid import uuid4
from datetime import datetime
#import random
import json
import numpy as np
#from icecream import install, ic
from rich import print as printanyway
from threading import Event
import os
from traceback import format_exc
from v2realbot.strategyblocks.activetrade.helpers import insert_SL_history
# - close means change status in prescribed Trends,update profit, delete from activeTrade
def close_position(state, data, direction: TradeDirection, reason: str, followup: Followup = None):
followup_text = str(followup) if followup is not None else ""
state.ilog(lvl=1,e=f"CLOSING TRADE {followup_text} {reason} {str(direction)}", curr_price=data["close"], trade=state.vars.activeTrade)
if direction == TradeDirection.SHORT:
res = state.buy(size=abs(int(state.positions)))
if isinstance(res, int) and res < 0:
raise Exception(f"error in required operation {reason} {res}")
elif direction == TradeDirection.LONG:
res = state.sell(size=state.positions)
if isinstance(res, int) and res < 0:
raise Exception(f"error in required operation STOPLOSS SELL {res}")
else:
raise Exception(f"unknow TradeDirection in close_position")
#pri uzavreni tradu zapisujeme SL history - lepsi zorbazeni v grafu
insert_SL_history(state)
state.vars.pending = state.vars.activeTrade.id
state.vars.activeTrade = None
state.vars.last_exit_index = data["index"]
if followup is not None:
state.vars.requested_followup = followup

172
v2realbot/strategyblocks/activetrade/close/conditions.py Normal file
View File

@ -0,0 +1,172 @@
from v2realbot.strategy.base import StrategyState
from v2realbot.strategy.StrategyOrderLimitVykladaciNormalizedMYSELL import StrategyOrderLimitVykladaciNormalizedMYSELL
from v2realbot.enums.enums import RecordType, StartBarAlign, Mode, Account, Followup
from v2realbot.common.PrescribedTradeModel import Trade, TradeDirection, TradeStatus
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.utils.directive_utils import get_conditions_from_configuration
from v2realbot.ml.mlutils import load_model
from v2realbot.common.model import SLHistory
from v2realbot.config import KW
#from icecream import install, ic
from rich import print as printanyway
from threading import Event
import os
from traceback import format_exc
from v2realbot.strategyblocks.indicators.helpers import evaluate_directive_conditions
from v2realbot.strategyblocks.activetrade.helpers import get_override_for_active_trade, normalize_tick
def dontexit_protection_met(state, data, direction: TradeDirection):
if direction == TradeDirection.LONG:
smer = "long"
else:
smer = "short"
mother_signal = state.vars.activeTrade.generated_by
if mother_signal is not None:
#TESTUJEME DONT_EXIT_
cond_dict = state.vars.conditions[KW.dont_exit][mother_signal][smer]
#OR
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "OR")
state.ilog(lvl=1,e=f"DONT_EXIT {mother_signal} {smer} =OR= {result}", **conditions_met, cond_dict=cond_dict)
if result:
return True
#OR neprosly testujeme AND
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "AND")
state.ilog(lvl=1,e=f"DONT_EXIT {mother_signal} {smer} =AND= {result}", **conditions_met, cond_dict=cond_dict)
if result:
return True
cond_dict = state.vars.conditions[KW.dont_exit]["common"][smer]
#OR
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "OR")
state.ilog(lvl=1,e=f"DONT_EXIT common {smer} =OR= {result}", **conditions_met, cond_dict=cond_dict)
if result:
return True
#OR neprosly testujeme AND
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "AND")
state.ilog(lvl=1,e=f"DONT_EXIT common {smer} =AND= {result}", **conditions_met, cond_dict=cond_dict)
return result
def exit_conditions_met(state, data, direction: TradeDirection):
if direction == TradeDirection.LONG:
smer = "long"
else:
smer = "short"
directive_name = "exit_cond_only_on_confirmed"
exit_cond_only_on_confirmed = get_override_for_active_trade(state, directive_name=directive_name, default_value=safe_get(state.vars, directive_name, False))
if exit_cond_only_on_confirmed and data['confirmed'] == 0:
state.ilog(lvl=0,e="EXIT COND ONLY ON CONFIRMED BAR")
return False
## minimální počet barů od vstupu
directive_name = "exit_cond_req_bars"
exit_cond_req_bars = get_override_for_active_trade(state, directive_name=directive_name, default_value=safe_get(state.vars, directive_name, 1))
if state.vars.last_in_index is not None:
index_to_compare = int(state.vars.last_in_index)+int(exit_cond_req_bars)
if int(data["index"]) < index_to_compare:
state.ilog(lvl=1,e=f"EXIT COND WAITING on required bars from IN {exit_cond_req_bars} TOO SOON", currindex=data["index"], index_to_compare=index_to_compare, last_in_index=state.vars.last_in_index)
return False
#POKUD je nastaven MIN PROFIT, zkontrolujeme ho a az pripadne pustime CONDITIONY
directive_name = "exit_cond_min_profit"
exit_cond_min_profit_nodir = get_override_for_active_trade(state, directive_name=directive_name, default_value=safe_get(state.vars, directive_name, None))
directive_name = "exit_cond_min_profit_" + str(smer)
exit_cond_min_profit = get_override_for_active_trade(state, directive_name=directive_name, default_value=exit_cond_min_profit_nodir)
#máme nastavený exit_cond_min_profit
# zjistíme, zda jsme v daném profit a případně nepustíme dál
# , zjistíme aktuální cenu a přičteme k avgp tento profit a podle toho pustime dal
if exit_cond_min_profit is not None:
exit_cond_min_profit_normalized = normalize_tick(state, data, float(exit_cond_min_profit))
exit_cond_goal_price = price2dec(float(state.avgp)+exit_cond_min_profit_normalized,3) if int(state.positions) > 0 else price2dec(float(state.avgp)-exit_cond_min_profit_normalized,3)
curr_price = float(data["close"])
state.ilog(lvl=1,e=f"EXIT COND min profit {exit_cond_goal_price=} {exit_cond_min_profit=} {exit_cond_min_profit_normalized=} {curr_price=}")
if (int(state.positions) < 0 and curr_price<=exit_cond_goal_price) or (int(state.positions) > 0 and curr_price>=exit_cond_goal_price):
state.ilog(lvl=1,e=f"EXIT COND min profit PASS - POKRACUJEME")
else:
state.ilog(lvl=1,e=f"EXIT COND min profit NOT PASS")
return False
#TOTO ZATIM NEMA VYZNAM
# options = safe_get(state.vars, 'exit_conditions', None)
# if options is None:
# state.ilog(lvl=0,e="No options for exit conditions in stratvars")
# return False
# disable_exit_proteciton_when = dict(AND=dict(), OR=dict())
# #preconditions
# disable_exit_proteciton_when['disabled_in_config'] = safe_get(options, 'enabled', False) is False
# #too good to be true (maximum profit)
# #disable_sell_proteciton_when['tgtbt_reached'] = safe_get(options, 'tgtbt', False) is False
# disable_exit_proteciton_when['disable_if_positions_above'] = int(safe_get(options, 'disable_if_positions_above', 0)) < abs(int(state.positions))
# #testing preconditions
# result, conditions_met = eval_cond_dict(disable_exit_proteciton_when)
# if result:
# state.ilog(lvl=0,e=f"EXIT_CONDITION for{smer} DISABLED by {conditions_met}", **conditions_met)
# return False
#bereme bud exit condition signalu, ktery activeTrade vygeneroval+ fallback na general
state.ilog(lvl=0,e=f"EXIT CONDITIONS ENTRY {smer}", conditions=state.vars.conditions[KW.exit])
mother_signal = state.vars.activeTrade.generated_by
if mother_signal is not None:
cond_dict = state.vars.conditions[KW.exit][state.vars.activeTrade.generated_by][smer]
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "OR")
state.ilog(lvl=1,e=f"EXIT CONDITIONS of {mother_signal} =OR= {result}", **conditions_met, cond_dict=cond_dict)
if result:
return True
#OR neprosly testujeme AND
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "AND")
state.ilog(lvl=1,e=f"EXIT CONDITIONS of {mother_signal} =AND= {result}", **conditions_met, cond_dict=cond_dict)
if result:
return True
#pokud nemame mother signal nebo exit nevratil nic, fallback na common
cond_dict = state.vars.conditions[KW.exit]["common"][smer]
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "OR")
state.ilog(lvl=1,e=f"EXIT CONDITIONS of COMMON =OR= {result}", **conditions_met, cond_dict=cond_dict)
if result:
return True
#OR neprosly testujeme AND
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "AND")
state.ilog(lvl=1,e=f"EXIT CONDITIONS of COMMON =AND= {result}", **conditions_met, cond_dict=cond_dict)
if result:
return True
#ZVAZIT JESTLI nesledujici puvodni pravidlo pro dontsellwhen pujdou realizovat inverzne jako exit when
#PUVODNI NASTAVENI - IDENTIFIKOVAce rustoveho MOMENTA - pokud je momentum, tak prodávat později
# #pokud je slope too high, pak prodavame jakmile slopeMA zacne klesat, napr. 4MA (TODO 3)
# #TODO zkusit pro pevny profit, jednoduse pozdrzet prodej - dokud tick_price roste nebo se drzi tak neprodavat, pokud klesne prodat
# #mozna mit dva mody - pri vetsi volatilite pouzivat momentum, pri mensi nebo kdyz potrebuju pryc, tak prodat hned
#puvodni nastaveni
#slopeMA_rising = 2
#rsi_not_falling = 3
# #toto docasne pryc dont_sell_when['slope_too_high'] = slope_too_high() and not isfalling(state.indicators.slopeMA,4)
# dont_sell_when['AND']['slopeMA_rising'] = isrising(state.indicators.slopeMA,safe_get(options, 'slopeMA_rising', 2))
# dont_sell_when['AND']['rsi_not_falling'] = not isfalling(state.indicators.RSI14,safe_get(options, 'rsi_not_falling',3))
# #dont_sell_when['rsi_dont_buy'] = state.indicators.RSI14[-1] > safe_get(state.vars, "rsi_dont_buy_above",50)
# result, conditions_met = eval_cond_dict(dont_sell_when)
# if result:
# state.ilog(lvl=0,e=f"SELL_PROTECTION {conditions_met} enabled")
# return result

147
v2realbot/strategyblocks/activetrade/close/evaluate_close.py Normal file
View File

@ -0,0 +1,147 @@
from v2realbot.strategyblocks.activetrade.close.close_position import close_position
from v2realbot.strategy.base import StrategyState
from v2realbot.enums.enums import Followup
from v2realbot.common.PrescribedTradeModel import Trade, TradeDirection, TradeStatus
from v2realbot.utils.utils import safe_get
from v2realbot.config import KW
#from icecream import install, ic
from rich import print as printanyway
from threading import Event
import os
from traceback import format_exc
from v2realbot.strategyblocks.activetrade.close.conditions import dontexit_protection_met, exit_conditions_met
from v2realbot.strategyblocks.activetrade.helpers import get_max_profit_price, get_profit_target_price, get_override_for_active_trade, keyword_conditions_met
def eval_close_position(state: StrategyState, data):
curr_price = float(data['close'])
state.ilog(lvl=0,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:
#close position handling
#TBD pridat OPTIMALIZACI POZICE - EXIT 1/2
#mame short pozice - (IDEA: rozlisovat na zaklade aktivniho tradu - umozni mi spoustet i pri soucasne long pozicemi)
if int(state.positions) < 0:
#get TARGET PRICE pro dany smer a signal
goal_price = get_profit_target_price(state, data, TradeDirection.SHORT)
max_price = get_max_profit_price(state, data, TradeDirection.SHORT)
state.ilog(lvl=1,e=f"Goal price {str(TradeDirection.SHORT)} {goal_price} max price {max_price}")
#EOD EXIT - TBD
#FORCED EXIT PRI KONCI DNE
#SL - execution
if curr_price > state.vars.activeTrade.stoploss_value:
directive_name = 'reverse_for_SL_exit_short'
reverse_for_SL_exit = get_override_for_active_trade(state=state, directive_name=directive_name, default_value=safe_get(state.vars, directive_name, "no"))
if reverse_for_SL_exit == "always":
followup_action = Followup.REVERSE
elif reverse_for_SL_exit == "cond":
followup_action = Followup.REVERSE if keyword_conditions_met(state, data, direction=TradeDirection.SHORT, keyword=KW.slreverseonly, skip_conf_validation=True) else None
else:
followup_action = None
close_position(state=state, data=data, direction=TradeDirection.SHORT, reason="SL REACHED", followup=followup_action)
return
#REVERSE BASED ON REVERSE CONDITIONS
if keyword_conditions_met(state, data, direction=TradeDirection.SHORT, keyword=KW.reverse):
close_position(state=state, data=data, direction=TradeDirection.SHORT, reason="REVERSE COND MET", followup=Followup.REVERSE)
return
#EXIT ADD CONDITIONS MET (exit and add)
if keyword_conditions_met(state, data, direction=TradeDirection.SHORT, keyword=KW.exitadd):
close_position(state=state, data=data, direction=TradeDirection.SHORT, reason="EXITADD COND MET", followup=Followup.ADD)
return
#CLOSING BASED ON EXIT CONDITIONS
if exit_conditions_met(state, data, TradeDirection.SHORT):
directive_name = 'reverse_for_cond_exit_short'
reverse_for_cond_exit_short = get_override_for_active_trade(state=state, directive_name=directive_name, default_value=safe_get(state.vars, directive_name, False))
directive_name = 'add_for_cond_exit_short'
add_for_cond_exit_short = get_override_for_active_trade(state=state, directive_name=directive_name, default_value=safe_get(state.vars, directive_name, False))
if reverse_for_cond_exit_short:
followup_action = Followup.REVERSE
elif add_for_cond_exit_short:
followup_action = Followup.ADD
else:
followup_action = None
close_position(state=state, data=data, direction=TradeDirection.SHORT, reason="EXIT COND MET", followup=followup_action)
return
#PROFIT
if curr_price<=goal_price:
#TODO cekat az slope prestane intenzivn erust, necekat az na klesani
#TODO mozna cekat na nejaky signal RSI
#TODO pripadne pokud dosahne TGTBB prodat ihned
max_price_signal = curr_price<=max_price
#OPTIMALIZACE pri stoupajícím angle
if max_price_signal or dontexit_protection_met(state=state, data=data,direction=TradeDirection.SHORT) is False:
close_position(state=state, data=data, direction=TradeDirection.SHORT, reason=f"PROFIT or MAXPROFIT REACHED {max_price_signal=}")
return
#mame long
elif int(state.positions) > 0:
#get TARGET PRICE pro dany smer a signal
goal_price = get_profit_target_price(state, data, TradeDirection.LONG)
max_price = get_max_profit_price(state, data, TradeDirection.LONG)
state.ilog(lvl=1,e=f"Goal price {str(TradeDirection.LONG)} {goal_price} max price {max_price}")
#EOD EXIT - TBD
#SL - execution
if curr_price < state.vars.activeTrade.stoploss_value:
directive_name = 'reverse_for_SL_exit_long'
reverse_for_SL_exit = get_override_for_active_trade(state, directive_name=directive_name, default_value=safe_get(state.vars, directive_name, "no"))
state.ilog(lvl=1, e=f"reverse_for_SL_exit {reverse_for_SL_exit}")
if reverse_for_SL_exit == "always":
followup_action = Followup.REVERSE
elif reverse_for_SL_exit == "cond":
followup_action = Followup.REVERSE if keyword_conditions_met(state, data, direction=TradeDirection.LONG, keyword=KW.slreverseonly, skip_conf_validation=True) else None
else:
followup_action = None
close_position(state=state, data=data, direction=TradeDirection.LONG, reason="SL REACHED", followup=followup_action)
return
#REVERSE BASED ON REVERSE CONDITIONS
if keyword_conditions_met(state, data,TradeDirection.LONG, KW.reverse):
close_position(state=state, data=data, direction=TradeDirection.LONG, reason="REVERSE COND MET", followup=Followup.REVERSE)
return
#EXIT ADD CONDITIONS MET (exit and add)
if keyword_conditions_met(state, data, TradeDirection.LONG, KW.exitadd):
close_position(state=state, data=data, direction=TradeDirection.LONG, reason="EXITADD COND MET", followup=Followup.ADD)
return
#EXIT CONDITIONS
if exit_conditions_met(state, data, TradeDirection.LONG):
directive_name = 'reverse_for_cond_exit_long'
reverse_for_cond_exit_long = get_override_for_active_trade(state, directive_name=directive_name, default_value=safe_get(state.vars, directive_name, False))
directive_name = 'add_for_cond_exit_long'
add_for_cond_exit_long = get_override_for_active_trade(state, directive_name=directive_name, default_value=safe_get(state.vars, directive_name, False))
if reverse_for_cond_exit_long:
followup_action = Followup.REVERSE
elif add_for_cond_exit_long:
followup_action = Followup.ADD
else:
followup_action = None
close_position(state=state, data=data, direction=TradeDirection.LONG, reason="EXIT CONDS MET", followup=followup_action)
return
#PROFIT
if curr_price>=goal_price:
#TODO cekat az slope prestane intenzivn erust, necekat az na klesani
#TODO mozna cekat na nejaky signal RSI
#TODO pripadne pokud dosahne TGTBB prodat ihned
max_price_signal = curr_price>=max_price
#OPTIMALIZACE pri stoupajícím angle
if max_price_signal or dontexit_protection_met(state, data, direction=TradeDirection.LONG) is False:
close_position(state=state, data=data, direction=TradeDirection.LONG, reason=f"PROFIT or MAXPROFIT REACHED {max_price_signal=}")
return

186
v2realbot/strategyblocks/activetrade/helpers.py Normal file
View File

@ -0,0 +1,186 @@
from v2realbot.strategy.base import StrategyState
from v2realbot.strategy.StrategyOrderLimitVykladaciNormalizedMYSELL import StrategyOrderLimitVykladaciNormalizedMYSELL
from v2realbot.enums.enums import RecordType, StartBarAlign, Mode, Account, Followup
from v2realbot.common.PrescribedTradeModel import Trade, TradeDirection, TradeStatus
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.utils.directive_utils import get_conditions_from_configuration
from v2realbot.ml.mlutils import load_model
from v2realbot.common.model import SLHistory
from v2realbot.config import KW
from uuid import uuid4
from datetime import datetime
#import random
import json
import numpy as np
#from icecream import install, ic
from rich import print as printanyway
from threading import Event
import os
from traceback import format_exc
from v2realbot.strategyblocks.helpers import normalize_tick
from v2realbot.strategyblocks.indicators.helpers import evaluate_directive_conditions
#otestuje keyword podminky (napr. reverse_if, nebo exitadd_if)
def keyword_conditions_met(state, data, direction: TradeDirection, keyword: KW, skip_conf_validation: bool = False):
action = str(keyword).upper()
if direction == TradeDirection.LONG:
smer = "long"
else:
smer = "short"
if skip_conf_validation is False:
directive_name = "exit_cond_only_on_confirmed"
exit_cond_only_on_confirmed = get_override_for_active_trade(state, directive_name=directive_name, default_value=safe_get(state.vars, directive_name, False))
if exit_cond_only_on_confirmed and data['confirmed'] == 0:
state.ilog(lvl=0,e=f"{action} CHECK COND ONLY ON CONFIRMED BAR")
return False
#TOTO zatim u REVERSU neresime
# #POKUD je nastaven MIN PROFIT, zkontrolujeme ho a az pripadne pustime CONDITIONY
# directive_name = "exit_cond_min_profit"
# exit_cond_min_profit = get_override_for_active_trade(directive_name=directive_name, default_value=safe_get(state.vars, directive_name, None))
# #máme nastavený exit_cond_min_profit
# # zjistíme, zda jsme v daném profit a případně nepustíme dál
# # , zjistíme aktuální cenu a přičteme k avgp tento profit a podle toho pustime dal
# if exit_cond_min_profit is not None:
# exit_cond_min_profit_normalized = normalize_tick(float(exit_cond_min_profit))
# exit_cond_goal_price = price2dec(float(state.avgp)+exit_cond_min_profit_normalized,3) if int(state.positions) > 0 else price2dec(float(state.avgp)-exit_cond_min_profit_normalized,3)
# curr_price = float(data["close"])
# state.ilog(lvl=0,e=f"EXIT COND min profit {exit_cond_goal_price=} {exit_cond_min_profit=} {exit_cond_min_profit_normalized=} {curr_price=}")
# if (int(state.positions) < 0 and curr_price<=exit_cond_goal_price) or (int(state.positions) > 0 and curr_price>=exit_cond_goal_price):
# state.ilog(lvl=0,e=f"EXIT COND min profit PASS - POKRACUJEME")
# else:
# state.ilog(lvl=0,e=f"EXIT COND min profit NOT PASS")
# return False
#TOTO ZATIM NEMA VYZNAM
# options = safe_get(state.vars, 'exit_conditions', None)
# if options is None:
# state.ilog(lvl=0,e="No options for exit conditions in stratvars")
# return False
# disable_exit_proteciton_when = dict(AND=dict(), OR=dict())
# #preconditions
# disable_exit_proteciton_when['disabled_in_config'] = safe_get(options, 'enabled', False) is False
# #too good to be true (maximum profit)
# #disable_sell_proteciton_when['tgtbt_reached'] = safe_get(options, 'tgtbt', False) is False
# disable_exit_proteciton_when['disable_if_positions_above'] = int(safe_get(options, 'disable_if_positions_above', 0)) < abs(int(state.positions))
# #testing preconditions
# result, conditions_met = eval_cond_dict(disable_exit_proteciton_when)
# if result:
# state.ilog(lvl=0,e=f"EXIT_CONDITION for{smer} DISABLED by {conditions_met}", **conditions_met)
# return False
#bereme bud exit condition signalu, ktery activeTrade vygeneroval+ fallback na general
state.ilog(lvl=0,e=f"{action} CONDITIONS ENTRY {smer}", conditions=state.vars.conditions[KW.reverse])
mother_signal = state.vars.activeTrade.generated_by
if mother_signal is not None:
cond_dict = state.vars.conditions[keyword][mother_signal][smer]
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "OR")
state.ilog(lvl=1,e=f"{action} CONDITIONS of {mother_signal} =OR= {result}", **conditions_met, cond_dict=cond_dict)
if result:
return True
#OR neprosly testujeme AND
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "AND")
state.ilog(lvl=1,e=f"{action} CONDITIONS of {mother_signal} =AND= {result}", **conditions_met, cond_dict=cond_dict)
if result:
return True
#pokud nemame mother signal nebo exit nevratil nic, fallback na common
cond_dict = state.vars.conditions[keyword]["common"][smer]
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "OR")
state.ilog(lvl=1,e=f"{action} CONDITIONS of COMMON =OR= {result}", **conditions_met, cond_dict=cond_dict)
if result:
return True
#OR neprosly testujeme AND
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "AND")
state.ilog(lvl=0,e=f"{action} CONDITIONS of COMMON =AND= {result}", **conditions_met, cond_dict=cond_dict)
if result:
return True
#mozna do SL helpers tuto
def insert_SL_history(state):
#insert stoploss history as key sl_history into runner archive extended data
state.extData["sl_history"].append(SLHistory(id=state.vars.activeTrade.id, time=state.time, sl_val=state.vars.activeTrade.stoploss_value))
def get_default_sl_value(state, direction: TradeDirection):
if direction == TradeDirection.LONG:
smer = "long"
else:
smer = "short"
#TODO zda signal, ktery activeTrade vygeneroval, nema vlastni nastaveni + fallback na general
options = safe_get(state.vars, 'exit', None)
if options is None:
state.ilog(lvl=1,e="No options for exit in stratvars. Fallback.")
return 0.01
directive_name = 'SL_defval_'+str(smer)
val = get_override_for_active_trade(state, directive_name=directive_name, default_value=safe_get(options, directive_name, 0.01))
return val
#funkce pro direktivy, ktere muzou byt overridnute v signal sekci
#tato funkce vyhleda signal sekci aktivniho tradu a pokusi se danou direktivu vyhledat tam,
#pokud nenajde tak vrati default, ktery byl poskytnut
def get_override_for_active_trade(state, directive_name: str, default_value: str):
val = default_value
override = "NO"
mother_signal = state.vars.activeTrade.generated_by
if mother_signal is not None:
override = "YES "+mother_signal
val = safe_get(state.vars.signals[mother_signal], directive_name, default_value)
state.ilog(lvl=0,e=f"{directive_name} OVERRIDE {override} NEWVAL:{val} ORIGINAL:{default_value} {mother_signal}", mother_signal=mother_signal,default_value=default_value)
return val
def get_profit_target_price(state, data, direction: TradeDirection):
if direction == TradeDirection.LONG:
smer = "long"
else:
smer = "short"
directive_name = "profit"
def_profit_both_directions = get_override_for_active_trade(state, directive_name=directive_name, default_value=safe_get(state.vars, directive_name, 0.50))
#profit pro dany smer
directive_name = 'profit_'+str(smer)
def_profit = get_override_for_active_trade(state, directive_name=directive_name, default_value=def_profit_both_directions)
normalized_def_profit = normalize_tick(state, data, float(def_profit))
state.ilog(lvl=0,e=f"PROFIT {def_profit=} {normalized_def_profit=}")
return price2dec(float(state.avgp)+normalized_def_profit,3) if int(state.positions) > 0 else price2dec(float(state.avgp)-normalized_def_profit,3)
def get_max_profit_price(state, data, direction: TradeDirection):
if direction == TradeDirection.LONG:
smer = "long"
else:
smer = "short"
directive_name = "max_profit"
max_profit_both_directions = get_override_for_active_trade(state, directive_name=directive_name, default_value=safe_get(state.vars, directive_name, 0.35))
#max profit pro dany smer, s fallbackem na bez smeru
directive_name = 'max_profit_'+str(smer)
max_profit = get_override_for_active_trade(state, directive_name=directive_name, default_value=max_profit_both_directions)
normalized_max_profit = normalize_tick(state,data,float(max_profit))
state.ilog(lvl=0,e=f"MAX PROFIT {max_profit=} {normalized_max_profit=}")
return price2dec(float(state.avgp)+normalized_max_profit,3) if int(state.positions) > 0 else price2dec(float(state.avgp)-normalized_max_profit,3)

89
v2realbot/strategyblocks/activetrade/sl/trailsl.py Normal file
View File

@ -0,0 +1,89 @@
from v2realbot.strategy.base import StrategyState
from v2realbot.common.PrescribedTradeModel import Trade, TradeDirection, TradeStatus
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get
#from icecream import install, ic
from rich import print as printanyway
from threading import Event
import os
from traceback import format_exc
from v2realbot.strategyblocks.activetrade.helpers import get_override_for_active_trade, normalize_tick, insert_SL_history
#pokud se cena posouva nasim smerem olespon o (0.05) nad (SL + 0.09val), posuneme SL o offset
#+ varianta - skoncit breakeven
#DIREKTIVY:
#maximalni stoploss, fallout pro "exit_short_if" direktivy
# SL_defval_short = 0.10
# SL_defval_long = 0.10
# SL_trailing_enabled_short = true
# SL_trailing_enabled_long = true
# #minimalni vzdalenost od aktualni SL, aby se SL posunula na
# SL_trailing_offset_short = 0.05
# SL_trailing_offset_long = 0.05
# #zda trailing zastavit na brakeeven
# SL_trailing_stop_at_breakeven_short = true
# SL_trailing_stop_at_breakeven_long = true
def trail_SL_management(state: StrategyState, data):
if int(state.positions) != 0 and float(state.avgp)>0 and state.vars.pending is None:
if int(state.positions) < 0:
direction = TradeDirection.SHORT
smer = "short"
else:
direction = TradeDirection.LONG
smer = "long"
# zatim nastaveni SL plati pro vsechny - do budoucna per signal - pridat sekci
options = safe_get(state.vars, 'exit', None)
if options is None:
state.ilog(lvl=1,e="Trail SL. No options for exit conditions in stratvars.")
return
directive_name = 'SL_trailing_enabled_'+str(smer)
sl_trailing_enabled = get_override_for_active_trade(state=state, directive_name=directive_name, default_value=safe_get(options, directive_name, False))
#SL_trailing_protection_window_short
directive_name = 'SL_trailing_protection_window_'+str(smer)
SL_trailing_protection_window = get_override_for_active_trade(state=state, directive_name=directive_name, default_value=safe_get(options, directive_name, 0))
index_to_compare = int(state.vars.last_in_index)+int(SL_trailing_protection_window)
if index_to_compare > int(data["index"]):
state.ilog(lvl=1,e=f"SL trail PROTECTION WINDOW {SL_trailing_protection_window} - TOO SOON", currindex=data["index"], index_to_compare=index_to_compare, last_in_index=state.vars.last_in_index)
return
if sl_trailing_enabled is True:
directive_name = 'SL_trailing_stop_at_breakeven_'+str(smer)
stop_breakeven = get_override_for_active_trade(state=state, directive_name=directive_name, default_value=safe_get(options, directive_name, False))
directive_name = 'SL_defval_'+str(smer)
def_SL = get_override_for_active_trade(state=state, directive_name=directive_name, default_value=safe_get(options, directive_name, 0.01))
directive_name = "SL_trailing_offset_"+str(smer)
offset = get_override_for_active_trade(state=state, directive_name=directive_name, default_value=safe_get(options, directive_name, 0.01))
#pokud je pozadovan trail jen do breakeven a uz prekroceno
if (direction == TradeDirection.LONG and stop_breakeven and state.vars.activeTrade.stoploss_value >= float(state.avgp)) or (direction == TradeDirection.SHORT and stop_breakeven and state.vars.activeTrade.stoploss_value <= float(state.avgp)):
state.ilog(lvl=1,e=f"SL trail STOP at breakeven {str(smer)} SL:{state.vars.activeTrade.stoploss_value} UNCHANGED", stop_breakeven=stop_breakeven)
return
#IDEA: Nyni posouvame SL o offset, mozna ji posunout jen o direktivu step ?
offset_normalized = normalize_tick(state, data, offset) #to ticks and from options
def_SL_normalized = normalize_tick(state, data, def_SL)
if direction == TradeDirection.LONG:
move_SL_threshold = state.vars.activeTrade.stoploss_value + offset_normalized + def_SL_normalized
state.ilog(lvl=1,e=f"SL TRAIL EVAL {smer} SL:{round(state.vars.activeTrade.stoploss_value,3)} TRAILGOAL:{move_SL_threshold}", def_SL=def_SL, offset=offset, offset_normalized=offset_normalized, def_SL_normalized=def_SL_normalized)
if (move_SL_threshold) < data['close']:
state.vars.activeTrade.stoploss_value += offset_normalized
insert_SL_history(state)
state.ilog(lvl=1,e=f"SL TRAIL TH {smer} reached {move_SL_threshold} SL moved to {state.vars.activeTrade.stoploss_value}", offset_normalized=offset_normalized, def_SL_normalized=def_SL_normalized)
elif direction == TradeDirection.SHORT:
move_SL_threshold = state.vars.activeTrade.stoploss_value - offset_normalized - def_SL_normalized
state.ilog(lvl=0,e=f"SL TRAIL EVAL {smer} SL:{round(state.vars.activeTrade.stoploss_value,3)} TRAILGOAL:{move_SL_threshold}", def_SL=def_SL, offset=offset, offset_normalized=offset_normalized, def_SL_normalized=def_SL_normalized)
if (move_SL_threshold) > data['close']:
state.vars.activeTrade.stoploss_value -= offset_normalized
insert_SL_history(state)
state.ilog(lvl=1,e=f"SL TRAIL GOAL {smer} reached {move_SL_threshold} SL moved to {state.vars.activeTrade.stoploss_value}", offset_normalized=offset_normalized, def_SL_normalized=def_SL_normalized)

47
v2realbot/strategyblocks/helpers.py Normal file
View File

@ -0,0 +1,47 @@
from v2realbot.strategy.base import StrategyState
from v2realbot.strategy.StrategyOrderLimitVykladaciNormalizedMYSELL import StrategyOrderLimitVykladaciNormalizedMYSELL
from v2realbot.enums.enums import RecordType, StartBarAlign, Mode, Account, Followup
from v2realbot.common.PrescribedTradeModel import Trade, TradeDirection, TradeStatus
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.utils.directive_utils import get_conditions_from_configuration
from v2realbot.ml.mlutils import load_model
from v2realbot.common.model import SLHistory
from v2realbot.config import KW
from uuid import uuid4
from datetime import datetime
#import random
#from icecream import install, ic
from rich import print as printanyway
from threading import Event
import os
from traceback import format_exc
def normalize_tick(state, data, tick: float, price: float = None, return_two_decimals: bool = False):
"""
Pokud je nastaveno v direktive:
#zda normalizovat vsechyn ticky (tzn. profit, maxprofit, SL atp.)
Normalize_ticks= true
Normalized Tick base price = 30
prevede normalizovany tick na tick odpovidajici vstupni cene
vysledek je zaokoruhleny na 2 des.mista
u cen pod 30, vrací 0.01. U cen nad 30 vrací pomerne zvetsene,
"""
#nemusime dodavat cenu, bereme aktualni
if price is None:
price = data["close"]
normalize_ticks = safe_get(state.vars, "normalize_ticks",False)
normalized_base_price = safe_get(state.vars, "normalized_base_price",30)
if normalize_ticks:
if price<normalized_base_price:
return tick
else:
#ratio of price vs base price
ratio = price/normalized_base_price
normalized_tick = ratio*tick
return price2dec(normalized_tick) if return_two_decimals else normalized_tick
else:
return tick

49
v2realbot/strategyblocks/indicators/RSI.py Normal file
View File

@ -0,0 +1,49 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from v2realbot.indicators.indicators import ema, natr, roc
from v2realbot.indicators.oscillators import rsi
from traceback import format_exc
#RSI INDICATOR
# type = RSI, source = [close, vwap, hlcc4], rsi_length = [14], MA_length = int (optional), on_confirmed_only = [true, false]
# pokud existuje MA, vytvarime i stejnojnojmenny MAcko
def populate_dynamic_RSI_indicator(data, state: StrategyState, name):
ind_type = "RSI"
options = safe_get(state.vars.indicators, name, None)
if options is None:
state.ilog(lvl=1,e=f"No options for {name} in stratvars")
return
if safe_get(options, "type", False) is False or safe_get(options, "type", False) != ind_type:
state.ilog(lvl=1,e="Type error")
return
#poustet kazdy tick nebo jenom na confirmed baru (on_confirmed_only = true)
on_confirmed_only = safe_get(options, 'on_confirmed_only', False)
req_source = safe_get(options, 'source', 'vwap')
if req_source not in ["close", "vwap","hlcc4"]:
state.ilog(lvl=1,e=f"Unknown source error {req_source} for {name}")
return
rsi_length = int(safe_get(options, "RSI_length",14))
rsi_MA_length = safe_get(options, "MA_length", None)
if on_confirmed_only is False or (on_confirmed_only is True and data['confirmed']==1):
try:
source = state.bars[req_source]
#cekame na dostatek dat
if len(source) > rsi_length:
rsi_res = rsi(source, rsi_length)
rsi_value = round(rsi_res[-1],4)
state.indicators[name][-1]=rsi_value
state.ilog(lvl=0,e=f"IND {name} RSI {rsi_value}")
if rsi_MA_length is not None:
src = state.indicators[name][-rsi_MA_length:]
rsi_MA_res = ema(src, rsi_MA_length)
rsi_MA_value = round(rsi_MA_res[-1],4)
state.indicators[name+"MA"][-1]=rsi_MA_value
state.ilog(lvl=0,e=f"IND {name} RSIMA {rsi_MA_value}")
else:
state.ilog(lvl=0,e=f"IND {name} RSI necháváme 0", message="not enough source data", source=source, rsi_length=rsi_length)
except Exception as e:
state.ilog(lvl=1,e=f"IND ERROR {name} RSI necháváme 0", message=str(e)+format_exc())

33
v2realbot/strategyblocks/indicators/atr.py Normal file
View File

@ -0,0 +1,33 @@
from v2realbot.indicators.indicators import ema, atr, roc
from v2realbot.indicators.oscillators import rsi
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from traceback import format_exc
#TODO ATR INDICATOR - predelat na CUSTOM a udelat scitani a odecteni od close (atru, atrd)
# type = ATR, ĺength = [14], on_confirmed_only = [true, false]
def populate_dynamic_atr_indicator(data, state: StrategyState, name):
ind_type = "ATR"
options = safe_get(state.vars.indicators, name, None)
if options is None:
state.ilog(lvl=1,e=f"No options for {name} in stratvars")
return
#poustet kazdy tick nebo jenom na confirmed baru (on_confirmed_only = true)
on_confirmed_only = safe_get(options, 'on_confirmed_only', False)
atr_length = int(safe_get(options, "length",5))
if on_confirmed_only is False or (on_confirmed_only is True and data['confirmed']==1):
try:
source_high = state.bars["high"][-atr_length:]
source_low = state.bars["low"][-atr_length:]
source_close = state.bars["close"][-atr_length:]
#if len(source) > ema_length:
atr_value = atr(source_high, source_low, source_close, atr_length)
val = round(atr_value[-1],4)
state.indicators[name][-1]= val
#state.indicators[name][-1]= round2five(val)
state.ilog(lvl=0,e=f"IND {name} ATR {val} {atr_length=}")
#else:
# state.ilog(lvl=0,e=f"IND {name} EMA necháváme 0", message="not enough source data", source=source, ema_length=ema_length)
except Exception as e:
state.ilog(lvl=0,e=f"IND ERROR {name} ATR necháváme 0", message=str(e)+format_exc())

23
v2realbot/strategyblocks/indicators/cbar_price.py Normal file
View File

@ -0,0 +1,23 @@
from v2realbot.strategy.base import StrategyState
def populate_cbar_tick_price_indicator(data, state: StrategyState):
try:
#pokud v potvrzovacím baru nebyly zmeny, nechavam puvodni hodnoty
# if tick_delta_volume == 0:
# state.indicators.tick_price[-1] = state.indicators.tick_price[-2]
# state.indicators.tick_volume[-1] = state.indicators.tick_volume[-2]
# else:
#tick_price = round2five(data['close'])
tick_price = data['close']
tick_delta_volume = data['volume'] - state.vars.last_tick_volume
state.cbar_indicators.tick_price[-1] = tick_price
state.cbar_indicators.tick_volume[-1] = tick_delta_volume
except:
pass
state.ilog(lvl=0,e=f"TICK PRICE {tick_price} VOLUME {tick_delta_volume} {data['confirmed']=}", prev_price=state.vars.last_tick_price, prev_volume=state.vars.last_tick_volume)
state.vars.last_tick_price = tick_price
state.vars.last_tick_volume = data['volume']

27
v2realbot/strategyblocks/indicators/cbar_rsi.py Normal file
View File

@ -0,0 +1,27 @@
from v2realbot.strategy.base import StrategyState
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from v2realbot.indicators.indicators import ema, natr, roc
from v2realbot.indicators.oscillators import rsi
from traceback import format_exc
#WIP
def populate_cbar_rsi_indicator(data, state):
#CBAR RSI indicator
options = safe_get(state.vars.indicators, 'crsi', None)
if options is None:
state.ilog(lvl=1,e="No options for crsi in stratvars")
return
try:
crsi_length = int(safe_get(options, 'crsi_length', 14))
source = state.cbar_indicators.tick_price #[-rsi_length:] #state.bars.vwap
crsi_res = rsi(source, crsi_length)
crsi_value = crsi_res[-1]
if str(crsi_value) == "nan":
crsi_value = 0
state.cbar_indicators.CRSI[-1]=crsi_value
#state.ilog(lvl=0,e=f"RSI {rsi_length=} {rsi_value=} {rsi_dont_buy=} {rsi_buy_signal=}", rsi_indicator=state.indicators.RSI14[-5:])
except Exception as e:
state.ilog(lvl=1,e=f"CRSI {crsi_length=} necháváme 0", message=str(e)+format_exc())
#state.indicators.RSI14[-1]=0

1
v2realbot/strategyblocks/indicators/custom/__init__.py Normal file
View File

@ -0,0 +1 @@
from . import *

72
v2realbot/strategyblocks/indicators/custom/_upscaled_rsi_wip.py Normal file
View File

@ -0,0 +1,72 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from v2realbot.indicators.indicators import ema, natr, roc
from v2realbot.strategyblocks.indicators.helpers import get_source_series
from rich import print as printanyway
from traceback import format_exc
from v2realbot.ml.ml import ModelML
import numpy as np
from collections import defaultdict
#WIP
#indicator to run on bar multiples
#např. umožní RSI na 5min
#params: resolution (bar multiples)
def upscaledrsi(state, params):
funcName = "upscaledrsi"
#new res in seconds
new_resolution = safe_get(params, "resolution", None)
old_resolution = state.bars["resolution"][-1]
#pokud potrebuju vsechny bary, tak si je dotahnu
new_bars = {}
new_bars = create_new_bars(state.bars, new_resolution)
#val = rsi(bars.)
#pokud potrebuju jen close nebo open muzu pouzit toto
# vezme to N-th element z pole
def resample_close_prices(bars, new_resolution):
# Check that the new resolution is a multiple of the old resolution.
if new_resolution % bars['resolution'][-1] != 0:
raise ValueError('New resolution must be a multiple of the old resolution.')
# Calculate the step size for selecting every Nth element.
step = new_resolution // bars['resolution'][-1]
# Extract close prices at the new resolution.
new_close_prices = bars['close'][::step]
#optimizied - but works only for numpy arrays, prevedeni z listu na numpy is costly - bars_array = np.array(bars)
#new_close_prices = np.take(bars['close'], np.arange(0, len(bars['close']), step), axis=0)
return new_close_prices
##TOTO PROJIT
#pokud je vstup jedna hodnota - muzu brat close,open v danem rozliseni tzn. jen N-th hodnotu zde
# Check that the new resolution is a multiple of the old resolution.
if new_resolution % state.bars["resolution"][-1] != 0:
raise ValueError('The new resolution must be a multiple of the old resolution.')
#get the number of bars in the new resolution.
n = new_resolution // old_resolution
# Calculate the new resolution values.
new_resolution_values = old_resolution_values.reshape((-1, new_resolution // len(old_resolution_values)))
# Select the N-th values from the new resolution values.
new_resolution_values[:, n]
source1 = safe_get(params, "source1", None)
if source1 in ["open","high","low","close","vwap","hlcc4"]:
source1_series = state.bars[source1]
else:
source1_series = state.indicators[source1]
source2 = safe_get(params, "source2", None)
if source2 in ["open","high","low","close","vwap","hlcc4"]:
source2_series = state.bars[source2]
else:
source2_series = state.indicators[source2]
mode = safe_get(params, "type")
state.ilog(lvl=0,e=f"INSIDE {funcName} {source1=} {source2=} {mode=}", **params)

23
v2realbot/strategyblocks/indicators/custom/barparams.py Normal file
View File

@ -0,0 +1,23 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from v2realbot.indicators.indicators import ema, natr, roc
from v2realbot.strategyblocks.indicators.helpers import get_source_series
from rich import print as printanyway
from traceback import format_exc
from v2realbot.ml.ml import ModelML
import numpy as np
from collections import defaultdict
#indicator allowing to be based on any bar parameter (index, high,open,close,trades,volume, etc.)
def barparams(state, params):
funcName = "barparams"
if params is None:
return -2, "params required"
source = safe_get(params, "source", None)
if source is None:
return -2, "source required"
try:
return 0, state.bars[source][-1]
except Exception as e:
return -2, str(e)+format_exc()

42
v2realbot/strategyblocks/indicators/custom/basestats.py Normal file
View File

@ -0,0 +1,42 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from v2realbot.indicators.indicators import ema, natr, roc
from v2realbot.strategyblocks.indicators.helpers import get_source_series
from rich import print as printanyway
from traceback import format_exc
from v2realbot.ml.ml import ModelML
import numpy as np
from collections import defaultdict
#vstupem je bud indicator nebo bar parametr
#na tomto vstupu dokaze provest zakladni statisticke funkce pro subpole X hodnot zpatky
#podporovane functions: min, max, mean
def basestats(state, params):
funcName = "basestats"
#name of indicator or
source = safe_get(params, "source", None)
lookback = safe_get(params, "lookback", None)
func = safe_get(params, "function", None)
source_dict = defaultdict(list)
source_dict[source] = get_source_series(state, source)
if lookback is None:
source_array = source_dict[source]
else:
try:
source_array = source_dict[source][-lookback-1:]
except IndexError:
source_array = source_dict[source]
if func == "min":
val = np.amin(source_array)
elif func == "max":
val = np.amax(source_array)
elif func == "mean":
val = np.mean(source_array)
else:
return -2, "wrong function"
return 0, val

58
v2realbot/strategyblocks/indicators/custom/conditional.py Normal file
View File

@ -0,0 +1,58 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from v2realbot.indicators.indicators import ema, natr, roc
from v2realbot.strategyblocks.indicators.helpers import get_source_series, evaluate_directive_conditions
from rich import print as printanyway
from traceback import format_exc
from v2realbot.ml.ml import ModelML
import numpy as np
from collections import defaultdict
#EXAMPLE of directives:
# [stratvars.indicators.novyconditional]
# type = "custom"
# subtype = "conditional"
# on_confirmed_only = true
# save_to_past = 5
# [stratvars.indicators.novyconditional.cp.conditions.isfalling]
# ema200.setindicator_if_falling = 3
# true_val = -1
# [stratvars.indicators.novyconditional.cp.conditions.isrising]
# ema200.setindicator_if_rising = 3
# true_val = 1
#novy podminkovy indikator, muze obsahovat az N podminek ve stejne syntaxy jako u signalu
#u kazde podminky je hodnota, ktera se vraci pokud je true
#hodi se pro vytvareni binarnich targetu pro ML
def conditional(state, params):
funcName = "conditional"
if params is None:
return -2, "params required"
conditions = safe_get(params, "conditions", None)
if conditions is None:
return -2, "conditions required"
try:
#workdict pro kazdou podminku se pripravi v initiu, v conditions mame pak novyatribut workdict
#muzeme mit vice podminek, ale prvni True vraci
for condname,condsettings in conditions.items():
#true davame jednicku default
true_val = safe_get(condsettings, "true_val", 1)
#printanyway(f"ind {name} podminka {condname} true_val {true_val}")
#zde je pripavena podminka, kterou jen evaluujeme
cond_dict = condsettings["cond_dict"]
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "OR")
state.ilog(lvl=1,e=f"IND PODMINKA {condname} =OR= {result}", **conditions_met, cond_dict=cond_dict)
if result:
return 0, true_val
#OR neprosly testujeme AND
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "AND")
state.ilog(lvl=1,e=f"IND PODMINKA {condname} =AND= {result}", **conditions_met, cond_dict=cond_dict)
if result:
return 0, true_val
return 0, 0
except Exception as e:
return -2, str(e)+format_exc()

181
v2realbot/strategyblocks/indicators/custom/custom_hub.py Normal file
View File

@ -0,0 +1,181 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from datetime import datetime, timedelta
from rich import print as printanyway
from v2realbot.indicators.indicators import ema
from traceback import format_exc
import importlib
#TODO TENTO IMPORT VYMYSLET, abych naloadoval package custom a nemusel nic pridat (vymyslet dynamicke volani z cele package ci)
#from v2realbot.strategyblocks.indicators.custom._upscaled_rsi_wip import upscaledrsi
from v2realbot.strategyblocks.indicators.custom.barparams import barparams
from v2realbot.strategyblocks.indicators.custom.basestats import basestats
from v2realbot.strategyblocks.indicators.custom.delta import delta
from v2realbot.strategyblocks.indicators.custom.divergence import divergence
from v2realbot.strategyblocks.indicators.custom.model import model
from v2realbot.strategyblocks.indicators.custom.opengap import opengap
from v2realbot.strategyblocks.indicators.custom.slope import slope
from v2realbot.strategyblocks.indicators.custom.conditional import conditional
from v2realbot.strategyblocks.indicators.custom.mathop import mathop
# import v2realbot.strategyblocks.indicators.custom as ci
def populate_dynamic_custom_indicator(data, state: StrategyState, name):
ind_type = "custom"
options = safe_get(state.vars.indicators, name, None)
if options is None:
state.ilog(lvl=1,e=f"No options for {name} in stratvars")
return
if safe_get(options, "type", False) is False or safe_get(options, "type", False) != ind_type:
state.ilog(lvl=1,e="Type error")
return
subtype = safe_get(options, 'subtype', False)
if subtype is False:
state.ilog(lvl=1,e=f"No subtype for {name} in stratvars")
return
#if MA is required
MA_length = safe_get(options, "MA_length", None)
active = safe_get(options, 'active', True)
if not active:
return
# např. 5 - znamená ulož hodnotu indikatoru 5 barů dozadu namísto posledni hodnoty - hodí se pro vytvareni targetu pro ML trening
save_to_past = int(safe_get(options, "save_to_past", 0))
def is_time_to_run():
# on_confirmed_only = true (def. False)
# start_at_bar_index = 2 (def. None)
# start_at_time = "9:31" (def. None)
# repeat_every_Nbar = N (def.None) (opakovat každý N bar, 1 - každý bar, 2 - každý 2., 0 - pouze jednou)
# repeat_every_Nmin = N (def. None) opakovat každých N minut
on_confirmed_only = safe_get(options, 'on_confirmed_only', False)
start_at_bar_index = safe_get(options, 'start_at_bar_index', None)
start_at_time = safe_get(options, 'start_at_time', None) # "9:30"
repeat_every_Nbar = safe_get(options, 'repeat_every_Nbar', None)
repeat_every_Nmin = safe_get(options, 'repeat_every_Nmin', None)
#stavové promenne v ramci indikatoru last_run_time a last_run_index - pro repeat_every.. direktivy
last_run_time = safe_get(options, 'last_run_time', None)
last_run_index = safe_get(options, 'last_run_index', None)
#confirmed
cond = on_confirmed_only is False or (on_confirmed_only is True and data['confirmed']==1)
if cond is False:
return cond, "not confirmed"
#start_at_time - v rámci optimalizace presunout do INIT parametru indikátorů, které se naplní v initu a celou dobu se nemění
if start_at_time is not None:
dt_now = datetime.fromtimestamp(data["updated"]).astimezone(zoneNY)
# Parse the maxTime string into a datetime object with the same date as timeA
req_start_time = datetime.strptime(start_at_time, "%H:%M").replace(
year=dt_now.year, month=dt_now.month, day=dt_now.day)
# Compare the time components (hours and minutes) of timeA and maxTime
if dt_now.time() > req_start_time.time():
state.ilog(lvl=0,e=f"IND {name} {subtype} START FROM TIME - PASSED: now:{dt_now.time()} reqtime:{req_start_time.time()}")
else:
state.ilog(lvl=0,e=f"IND {name} {subtype} START FROM TIME - NOT YET: now:{dt_now.time()} reqtime:{req_start_time.time()}")
cond = False
if cond is False:
return cond, "start_at_time not yet"
#start_on_bar = 0
if start_at_bar_index is not None:
cond = start_at_bar_index < data["index"]
if cond:
state.ilog(lvl=0,e=f"IND {name} {subtype} START FROM BAR - PASSED: now:{data['index']} reqbar:{start_at_bar_index}")
else:
state.ilog(lvl=0,e=f"IND {name} {subtype} START FROM BAR - NOT YET: now:{data['index']} reqbar:{start_at_bar_index}")
if cond is False:
return cond, "start_at_bar_index not yet"
#pokud 0 - opakujeme jednou, pokud 1 tak opakujeme vzdy, jinak dle poctu
if repeat_every_Nbar is not None:
#jiz bezelo - delame dalsi checky, pokud nebezelo, poustime jako true
if last_run_index is not None:
required_bar_to_run = last_run_index + repeat_every_Nbar
if repeat_every_Nbar == 0:
state.ilog(lvl=0,e=f"IND {name} {subtype} RUN ONCE ALREADY at:{last_run_index} at:{last_run_time}", repeat_every_Nbar=repeat_every_Nbar, last_run_index=last_run_index)
cond = False
elif repeat_every_Nbar == 1:
pass
elif data["index"] < required_bar_to_run:
state.ilog(lvl=0,e=f"IND {name} {subtype} REPEAT EVERY N BAR WAITING: req:{required_bar_to_run} now:{data['index']}", repeat_every_Nbar=repeat_every_Nbar, last_run_index=last_run_index)
cond = False
if cond is False:
return cond, "repeat_every_Nbar not yet"
#pokud nepozadovano, pak poustime
if repeat_every_Nmin is not None:
#porovnavame jen pokud uz bezelo
if last_run_time is not None:
required_time_to_run = last_run_time + timedelta(minutes=repeat_every_Nmin)
datetime_now = datetime.fromtimestamp(data["updated"]).astimezone(zoneNY)
if datetime_now < required_time_to_run:
state.ilog(lvl=0,e=f"IND {name} {subtype} REPEAT EVERY {repeat_every_Nmin}MINS WAITING", last_run_time=last_run_time, required_time_to_run=required_time_to_run, datetime_now=datetime_now)
cond = False
if cond is False:
return cond, "repeat_every_Nmin not yet"
return cond, "ok"
should_run, msg = is_time_to_run()
if should_run:
#TODO get custom params
custom_params = safe_get(options, "cp", None)
#vyplnime last_run_time a last_run_index
state.vars.indicators[name]["last_run_time"] = datetime.fromtimestamp(data["updated"]).astimezone(zoneNY)
state.vars.indicators[name]["last_run_index"] = data["index"]
# - volame custom funkci pro ziskani hodnoty indikatoru
# - tu ulozime jako novou hodnotu indikatoru a prepocteme MAcka pokud je pozadovane
# - pokud cas neni, nechavame puvodni, vcetna pripadneho MAcka
#pozor jako defaultní hodnotu dává engine 0 - je to ok?
try:
#subtype = "ci."+subtype
custom_function = eval(subtype)
res_code, new_val = custom_function(state, custom_params)
if res_code == 0:
state.indicators[name][-1-save_to_past]=new_val
state.ilog(lvl=1,e=f"IND {name} {subtype} VAL FROM FUNCTION: {new_val}", lastruntime=state.vars.indicators[name]["last_run_time"], lastrunindex=state.vars.indicators[name]["last_run_index"], save_to_past=save_to_past)
#prepocitame MA if required
if MA_length is not None:
src = state.indicators[name][-MA_length:]
MA_res = ema(src, MA_length)
MA_value = round(MA_res[-1],7)
state.indicators[name+"MA"][-1-save_to_past]=MA_value
state.ilog(lvl=0,e=f"IND {name}MA {subtype} {MA_value}",save_to_past=save_to_past)
else:
err = f"IND ERROR {name} {subtype}Funkce {custom_function} vratila {res_code} {new_val}."
raise Exception(err)
except Exception as e:
if len(state.indicators[name]) >= 2:
state.indicators[name][-1]=state.indicators[name][-2]
if MA_length is not None and len(state.indicators[name+"MA"])>=2:
state.indicators[name+"MA"][-1]=state.indicators[name+"MA"][-2]
state.ilog(lvl=1,e=f"IND ERROR {name} {subtype} necháváme původní", message=str(e)+format_exc())
else:
state.ilog(lvl=0,e=f"IND {name} {subtype} COND NOT READY: {msg}")
#not time to run
if len(state.indicators[name]) >= 2:
state.indicators[name][-1]=state.indicators[name][-2]
if MA_length is not None and len(state.indicators[name+"MA"])>=2:
state.indicators[name+"MA"][-1]=state.indicators[name+"MA"][-2]
state.ilog(lvl=0,e=f"IND {name} {subtype} NOT TIME TO RUN - value(and MA) still original")

24
v2realbot/strategyblocks/indicators/custom/delta.py Normal file
View File

@ -0,0 +1,24 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from v2realbot.indicators.indicators import ema, natr, roc
from v2realbot.strategyblocks.indicators.helpers import get_source_series
from rich import print as printanyway
from traceback import format_exc
from v2realbot.ml.ml import ModelML
import numpy as np
from collections import defaultdict
#strength, absolute change of parameter between current value and lookback value (n-past)
#used for example to measure unusual peaks
def delta(state, params):
funcName = "delta"
source = safe_get(params, "source", None)
lookback = safe_get(params, "lookback",1)
source_series = get_source_series(state, source)
lookbackval = source_series[-lookback-1]
currval = source_series[-1]
delta = currval - lookbackval
state.ilog(lvl=1,e=f"INSIDE {funcName} {delta} {source=} {lookback=}", currval=currval, lookbackval=lookbackval, **params)
return 0, delta

43
v2realbot/strategyblocks/indicators/custom/divergence.py Normal file
View File

@ -0,0 +1,43 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from v2realbot.indicators.indicators import ema, natr, roc
from v2realbot.strategyblocks.indicators.helpers import get_source_series
from rich import print as printanyway
from traceback import format_exc
from v2realbot.ml.ml import ModelML
import numpy as np
from collections import defaultdict
#abs/rel divergence of two indicators
def divergence(state, params):
funcName = "indicatorDivergence"
source1 = safe_get(params, "source1", None)
source1_series = get_source_series(state, source1)
source2 = safe_get(params, "source2", None)
source2_series = get_source_series(state, source2)
mode = safe_get(params, "type")
state.ilog(lvl=0,e=f"INSIDE {funcName} {source1=} {source2=} {mode=}", **params)
val = 0
if mode == "abs":
val = round(abs(float(source1_series[-1]) - float(source2_series[-1])),4)
elif mode == "absn":
val = round((abs(float(source1_series[-1]) - float(source2_series[-1])))/float(source1_series[-1]),4)
elif mode == "rel":
val = round(float(source1_series[-1]) - float(source2_series[-1]),4)
elif mode == "reln":
val = round((float(source1_series[-1]) - float(source2_series[-1]))/float(source1_series[-1]),4)
elif mode == "pctabs":
val = pct_diff(num1=float(source1_series[-1]),num2=float(source2_series[-1]), absolute=True)
elif mode == "pct":
val = pct_diff(num1=float(source1_series[-1]),num2=float(source2_series[-1]))
return 0, val
#model - naloadovana instance modelu
#seq - sekvence pro vstup

31
v2realbot/strategyblocks/indicators/custom/mathop.py Normal file
View File

@ -0,0 +1,31 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from v2realbot.strategyblocks.indicators.helpers import get_source_series, value_or_indicator
#allows basic mathematical operators to one or more indicators (add two indicator, add value to a indicator etc.)
def mathop(state, params):
funcName = "mathop"
#indicator name
source1 = safe_get(params, "source1", None)
source1_series = get_source_series(state, source1)
#indicator or value
source2 = safe_get(params, "source2", None)
operator = safe_get(params, "operator", None)
#state.ilog(lvl=0,e=f"INSIDE {funcName} {source1=} {source2=}", **params)
if source1 is None or source2 is None or operator is None:
return -2, "required source1 source2 operator"
if operator == "+":
val = round(float(source1_series[-1] + value_or_indicator(state, source2)),4)
elif operator == "-":
val = round(float(source1_series[-1] - value_or_indicator(state, source2)),4)
else:
return -2, "unknow operator"
#state.ilog(lvl=0,e=f"INSIDE {funcName} {source1=} {source2=} {val}", **params)
return 0, val

58
v2realbot/strategyblocks/indicators/custom/model.py Normal file
View File

@ -0,0 +1,58 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from v2realbot.indicators.indicators import ema, natr, roc
from v2realbot.strategyblocks.indicators.helpers import get_source_series
from rich import print as printanyway
from traceback import format_exc
from v2realbot.ml.ml import ModelML
import numpy as np
from collections import defaultdict
def model(state, params):
funcName = "model"
if params is None:
return -2, "params required"
name = safe_get(params, "name", None)
version = safe_get(params, "version", None)
#TBD co s temito, kdyz se budou brat z uloženého modelu?
#mozna jen na TRAIN?
# seq = safe_get(params, "seq", None)
# use_bars = safe_get(params, "use_bars", True)
# bar_features = safe_get(params, "bar_features", None)
# ind_features = safe_get(params, "ind_features", None)
# if name is None or ind_features is None:
# return -2, "name/ind_features required"
if not name in state.vars.loaded_models:
return -2, "model not loaded"
try:
mdl = state.vars.loaded_models[name]
if len(state.bars["close"]) < mdl.input_sequences:
return 0, 0
#return -2, f"too soon - not enough data for seq {seq=}"
value = mdl.predict(state.bars, state.indicators)
return 0, value
except Exception as e:
printanyway(str(e)+format_exc())
return -2, str(e)+format_exc()
#presunuto do classy modelu - DECOMISSIONOVAT
# def get_model_prediction(cfg: ModelML):
# lastNbars = slice_dict_lists(state.bars, cfg.seq, True)
# lastNindicators = slice_dict_lists(state.indicators, cfg.seq, False)
# combined_live_data = cfg.column_stack_source(lastNbars, lastNindicators)
# combined_live_data = cfg.scalerX.transform(combined_live_data)
# combined_live_data = np.array(combined_live_data)
# #converts to 3D array
# # 1 number of samples in the array.
# # 2 represents the sequence length.
# # 3 represents the number of features in the data.
# combined_live_data = combined_live_data.reshape((1, cfg.seq, combined_live_data.shape[1]))
# #prediction = model.predict(combined_live_data, verbose=0)
# prediction = cfg.model(combined_live_data, training=False)
# # Convert the prediction back to the original scale
# return float(cfg.scalerY.inverse_transform(prediction))

22
v2realbot/strategyblocks/indicators/custom/opengap.py Normal file
View File

@ -0,0 +1,22 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff
from v2realbot.strategy.base import StrategyState
from v2realbot.indicators.indicators import ema, natr, roc
from v2realbot.strategyblocks.indicators.helpers import get_source_series
from rich import print as printanyway
from traceback import format_exc
from v2realbot.ml.ml import ModelML
import numpy as np
from collections import defaultdict
#WIP -
#testing custom indicator CODE
def opengap(state, params):
funcName = "opengap"
param1 = safe_get(params, "param1")
param2 = safe_get(params, "param2")
state.ilog(lvl=0,e=f"INSIDE {funcName} {param1=} {param2=}", **params)
last_close = 28.45
today_open = 29.45
val = pct_diff(last_close, today_open)
return 0, val
#random.randint(10, 20)

35
v2realbot/strategyblocks/indicators/custom/slope.py Normal file
View File

@ -0,0 +1,35 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from v2realbot.indicators.indicators import ema, natr, roc
from v2realbot.strategyblocks.indicators.helpers import get_source_series
from rich import print as printanyway
from traceback import format_exc
from v2realbot.ml.ml import ModelML
import numpy as np
from collections import defaultdict
#rate of change - last value of source indicator vs lookback value of lookback_priceline indicator
def slope(state, params):
funcName = "slope"
source = safe_get(params, "source", None)
source_series = get_source_series(state, source)
lookback = safe_get(params, "lookback", 5)
lookback_priceline = safe_get(params, "lookback_priceline", None)
lookback_series = get_source_series(state, lookback_priceline)
try:
lookbackprice = lookback_series[-lookback-1]
lookbacktime = state.bars.updated[-lookback-1]
except IndexError:
max_delka = len(lookback_series)
lookbackprice =lookback_series[-max_delka]
lookbacktime = state.bars.updated[-max_delka]
#výpočet úhlu - a jeho normalizace
currval = source_series[-1]
slope = ((currval - lookbackprice)/abs(lookbackprice))*100
#slope = round(slope, 4)
state.ilog(lvl=1,e=f"INSIDE {funcName} {slope} {source=} {lookback=}", currval_source=currval, lookbackprice=lookbackprice, lookbacktime=lookbacktime, **params)
return 0, slope

38
v2realbot/strategyblocks/indicators/ema.py Normal file
View File

@ -0,0 +1,38 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from v2realbot.indicators.indicators import ema, natr, roc
from traceback import format_exc
#EMA INDICATOR
# type = EMA, source = [close, vwap, hlcc4], length = [14], on_confirmed_only = [true, false]
def populate_dynamic_ema_indicator(data, state: StrategyState, name):
ind_type = "EMA"
options = safe_get(state.vars.indicators, name, None)
if options is None:
state.ilog(lvl=1,e=f"No options for {name} in stratvars")
return
if safe_get(options, "type", False) is False or safe_get(options, "type", False) != ind_type:
state.ilog(lvl=1,e="Type error")
return
#poustet kazdy tick nebo jenom na confirmed baru (on_confirmed_only = true)
on_confirmed_only = safe_get(options, 'on_confirmed_only', False)
req_source = safe_get(options, 'source', 'vwap')
if req_source not in ["close", "vwap","hlcc4"]:
state.ilog(lvl=1,e=f"Unknown source error {req_source} for {name}")
return
ema_length = int(safe_get(options, "length",14))
if on_confirmed_only is False or (on_confirmed_only is True and data['confirmed']==1):
try:
source = state.bars[req_source][-ema_length:]
#if len(source) > ema_length:
ema_value = ema(source, ema_length)
val = round(ema_value[-1],4)
state.indicators[name][-1]= val
#state.indicators[name][-1]= round2five(val)
state.ilog(lvl=0,e=f"IND {name} EMA {val} {ema_length=}")
#else:
# state.ilog(lvl=0,e=f"IND {name} EMA necháváme 0", message="not enough source data", source=source, ema_length=ema_length)
except Exception as e:
state.ilog(lvl=1,e=f"IND ERROR {name} EMA necháváme 0", message=str(e)+format_exc())

87
v2realbot/strategyblocks/indicators/helpers.py Normal file
View File

@ -0,0 +1,87 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from traceback import format_exc
#ZATIM tyto zkopirovany SEM DO HELPERS
#podle toho jak se osvedci se zakl.indikatory to s state
#zatim se mi to moc nezda
def value_or_indicator(state,value):
#preklad direktivy podle typu, pokud je int anebo float - je to primo hodnota
#pokud je str, jde o indikator a dotahujeme posledni hodnotu z nej
if isinstance(value, (int, float)):
return value
elif isinstance(value, str):
try:
#pokud existuje v indikatoru MA bereme MA jinak indikator, pokud neexistuje bereme bar
ret = get_source_or_MA(state, indicator=value)[-1]
state.ilog(lvl=0,e=f"Pro porovnani bereme posledni hodnotu {ret} z indikatoru {value}")
except Exception as e :
ret = 0
state.ilog(lvl=1,e=f"Neexistuje indikator s nazvem {value} vracime 0" + str(e) + format_exc())
return ret
#OPTIMALIZOVANO CHATGPT
#funkce vytvori podminky (bud pro AND/OR) z pracovniho dict
def evaluate_directive_conditions(state, work_dict, cond_type):
def rev(kw, condition):
if directive.endswith(kw):
return not condition
else:
return condition
cond = {}
cond[cond_type] = {}
# Create a dictionary to map directives to functions
directive_functions = {
"above": lambda ind, val: get_source_or_MA(state, ind)[-1] > value_or_indicator(state,val),
"equals": lambda ind, val: get_source_or_MA(state, ind)[-1] == value_or_indicator(state,val),
"below": lambda ind, val: get_source_or_MA(state, ind)[-1] < value_or_indicator(state,val),
"falling": lambda ind, val: isfalling(get_source_or_MA(state, ind), val),
"rising": lambda ind, val: isrising(get_source_or_MA(state, ind), val),
"crossed_down": lambda ind, val: buy_if_crossed_down(state, ind, value_or_indicator(state,val)),
"crossed_up": lambda ind, val: buy_if_crossed_up(state, ind, value_or_indicator(state,val)),
"crossed": lambda ind, val: buy_if_crossed_down(state, ind, value_or_indicator(state,val)) or buy_if_crossed_up(state, ind, value_or_indicator(state,val)),
"pivot_a": lambda ind, val: is_pivot(source=get_source_or_MA(state, ind), leg_number=val, type="A"),
"pivot_v": lambda ind, val: is_pivot(source=get_source_or_MA(state, ind), leg_number=val, type="V"),
"still_for": lambda ind, val: is_still(get_source_or_MA(state, ind), val, 2),
}
for indname, directive, value in work_dict[cond_type]:
for keyword, func in directive_functions.items():
if directive.endswith(keyword):
cond[cond_type][directive + "_" + indname + "_" + str(value)] = rev("not_" + keyword, func(indname, value))
return eval_cond_dict(cond)
def get_source_or_MA(state, indicator):
#pokud ma, pouzije MAcko, pokud ne tak standardni indikator
#pokud to jmeno neexistuje, tak pripadne bere z barů (close,open,hlcc4, vwap atp.)
try:
return state.indicators[indicator+"MA"]
except KeyError:
try:
return state.indicators[indicator]
except KeyError:
return state.bars[indicator]
def get_source_series(state, source):
try:
return state.bars[source]
except KeyError:
return state.indicators[source]
#TYTO NEJSPIS DAT do util
#vrati true pokud dany indikator prekrocil threshold dolu
def buy_if_crossed_down(state, indicator, value):
res = crossed_down(threshold=value, list=get_source_or_MA(state, indicator))
#state.ilog(lvl=0,e=f"signal_if_crossed_down {indicator} {value} {res}")
return res
#vrati true pokud dany indikator prekrocil threshold nahoru
def buy_if_crossed_up(state, indicator, value):
res = crossed_up(threshold=value, list=get_source_or_MA(state, indicator))
#state.ilog(lvl=0,e=f"signal_if_crossed_up {indicator} {value} {res}")
return res

100
v2realbot/strategyblocks/indicators/indicators_hub.py Normal file
View File

@ -0,0 +1,100 @@
from v2realbot.strategy.base import StrategyState
from v2realbot.strategyblocks.indicators.cbar_price import populate_cbar_tick_price_indicator
from v2realbot.strategyblocks.indicators.custom.custom_hub import populate_dynamic_custom_indicator
from v2realbot.strategyblocks.indicators.slope import populate_dynamic_slope_indicator
from v2realbot.strategyblocks.indicators.slopeLP import populate_dynamic_slopeLP_indicator
from v2realbot.strategyblocks.indicators.ema import populate_dynamic_ema_indicator
from v2realbot.strategyblocks.indicators.RSI import populate_dynamic_RSI_indicator
from v2realbot.strategyblocks.indicators.natr import populate_dynamic_natr_indicator
from v2realbot.strategyblocks.indicators.atr import populate_dynamic_atr_indicator
import numpy as np
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
import json
def populate_all_indicators(data, state: StrategyState):
#TYTO MOZNA TAKY POSUNOUT OUT
def get_last_ind_vals():
last_ind_vals = {}
#print(state.indicators.items())
for key in state.indicators:
if key != 'time':
last_ind_vals[key] = state.indicators[key][-6:]
for key in state.cbar_indicators:
if key != 'time':
last_ind_vals[key] = state.cbar_indicators[key][-6:]
# for key in state.secondary_indicators:
# if key != 'time':
# last_ind_vals[key] = state.secondary_indicators[key][-5:]
return last_ind_vals
#zobrazí jak daleko od sebe chodí updaty (skupiny tradů co mění cenu) a průměr za 50jejich
def process_delta():
last_update_delta = round((float(data['updated']) - state.vars.last_update_time),6) if state.vars.last_update_time != 0 else 0
state.vars.last_update_time = float(data['updated'])
if len(state.vars.last_50_deltas) >=50:
state.vars.last_50_deltas.pop(0)
state.vars.last_50_deltas.append(last_update_delta)
avg_delta = np.mean(state.vars.last_50_deltas)
return last_update_delta, avg_delta
conf_bar = data['confirmed']
last_update_delta, avg_delta = process_delta()
state.ilog(lvl=1,e=f"-----{data['index']}-{conf_bar}--delta:{last_update_delta}---AVGdelta:{avg_delta}", data=data)
#kroky pro CONFIRMED BAR only
if conf_bar == 1:
#logika pouze pro potvrzeny bar
state.ilog(lvl=0,e="BAR potvrzeny")
#pri potvrzem CBARu nulujeme counter volume pro tick based indicator
state.vars.last_tick_volume = 0
state.vars.next_new = 1
#kroky pro CONTINOUS TICKS only
else:
#CBAR INDICATOR pro tick price a deltu VOLUME
populate_cbar_tick_price_indicator(data, state)
#TBD nize predelat na typizovane RSI (a to jak na urovni CBAR tak confirmed)
#populate_cbar_rsi_indicator()
#populate indicators, that have type in stratvars.indicators
populate_dynamic_indicators(data, state)
lp = data['close']
#TODO na toto se podivam, nejak moc zajasonovani a zpatky
#PERF PROBLEM
state.ilog(lvl=1,e="ENTRY", msg=f"LP:{lp} P:{state.positions}/{round(float(state.avgp),3)} SL:{state.vars.activeTrade.stoploss_value if state.vars.activeTrade is not None else None} profit:{round(float(state.profit),2)} Trades:{len(state.tradeList)} pend:{state.vars.pending}", activeTrade=json.loads(json.dumps(state.vars.activeTrade, default=json_serial)), prescribedTrades=json.loads(json.dumps(state.vars.prescribedTrades, default=json_serial)), pending=str(state.vars.pending))
inds = get_last_ind_vals()
state.ilog(lvl=1,e="Indikatory", **inds)
def populate_dynamic_indicators(data, state: StrategyState):
#pro vsechny indikatory, ktere maji ve svych stratvars TYPE, poustime populaci daneho typu indikaotru
for indname, indsettings in state.vars.indicators.items():
for option,value in indsettings.items():
if option == "type":
if value == "slope":
populate_dynamic_slope_indicator(data, state, name = indname)
#slope variant with continuous Left Point
elif value == "slopeLP":
populate_dynamic_slopeLP_indicator(data, state, name = indname)
elif value == "RSI":
populate_dynamic_RSI_indicator(data, state, name = indname)
elif value == "EMA":
populate_dynamic_ema_indicator(data, state, name = indname)
elif value == "NATR":
populate_dynamic_natr_indicator(data, state, name = indname)
elif value == "ATR":
populate_dynamic_atr_indicator(data, state, name = indname)
elif value == "custom":
populate_dynamic_custom_indicator(data, state, name = indname)

33
v2realbot/strategyblocks/indicators/natr.py Normal file
View File

@ -0,0 +1,33 @@
from v2realbot.indicators.indicators import ema, natr, roc
from v2realbot.indicators.oscillators import rsi
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from traceback import format_exc
#NATR INDICATOR
# type = NATR, ĺength = [14], on_confirmed_only = [true, false]
def populate_dynamic_natr_indicator(data, state: StrategyState, name):
ind_type = "NATR"
options = safe_get(state.vars.indicators, name, None)
if options is None:
state.ilog(lvl=1,e=f"No options for {name} in stratvars")
return
#poustet kazdy tick nebo jenom na confirmed baru (on_confirmed_only = true)
on_confirmed_only = safe_get(options, 'on_confirmed_only', False)
natr_length = int(safe_get(options, "length",5))
if on_confirmed_only is False or (on_confirmed_only is True and data['confirmed']==1):
try:
source_high = state.bars["high"][-natr_length:]
source_low = state.bars["low"][-natr_length:]
source_close = state.bars["close"][-natr_length:]
#if len(source) > ema_length:
natr_value = natr(source_high, source_low, source_close, natr_length)
val = round(natr_value[-1],4)
state.indicators[name][-1]= val
#state.indicators[name][-1]= round2five(val)
state.ilog(lvl=0,e=f"IND {name} NATR {val} {natr_length=}")
#else:
# state.ilog(lvl=0,e=f"IND {name} EMA necháváme 0", message="not enough source data", source=source, ema_length=ema_length)
except Exception as e:
state.ilog(lvl=0,e=f"IND ERROR {name} NATR necháváme 0", message=str(e)+format_exc())

132
v2realbot/strategyblocks/indicators/slope.py Normal file
View File

@ -0,0 +1,132 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from v2realbot.indicators.indicators import ema, natr, roc
from v2realbot.indicators.oscillators import rsi
import numpy as np
from traceback import format_exc
def populate_dynamic_slope_indicator(data, state: StrategyState, name):
options = safe_get(state.vars.indicators, name, None)
if options is None:
state.ilog(lvl=1,e="No options for slow slope in stratvars")
return
if safe_get(options, "type", False) is False or safe_get(options, "type", False) != "slope":
state.ilog(lvl=1,e="Type error")
return
#poustet kazdy tick nebo jenom na confirmed baru (on_confirmed_only = true)
on_confirmed_only = safe_get(options, 'on_confirmed_only', False)
#SLOW SLOPE INDICATOR
#úhel stoupání a klesání vyjádřený mezi -1 až 1
#pravý bod přímky je aktuální cena, levý je průměr X(lookback offset) starších hodnot od slope_lookback.
#VYSTUPY: state.indicators[name],
# state.indicators[nameMA]
# statický indikátor (angle) - stejneho jmena pro vizualizaci uhlu
if on_confirmed_only is False or (on_confirmed_only is True and data['confirmed']==1):
try:
slope_lookback = safe_get(options, 'slope_lookback', 100)
lookback_priceline = safe_get(options, 'lookback_priceline', None)
lookback_offset = safe_get(options, 'lookback_offset', 25)
minimum_slope = safe_get(options, 'minimum_slope', 25)
maximum_slope = safe_get(options, "maximum_slope",0.9)
#jako levy body pouzivame lookback_priceline INDIKATOR vzdaleny slope_lookback barů
if lookback_priceline is not None:
try:
lookbackprice = state.indicators[lookback_priceline][-slope_lookback-1]
lookbacktime = state.bars.updated[-slope_lookback-1]
except IndexError:
max_delka = len(state.indicators[lookback_priceline])
lookbackprice = state.indicators[lookback_priceline][-max_delka]
lookbacktime = state.bars.updated[-max_delka]
else:
#NEMAME LOOKBACK PRICLINE - pouzivame stary způsob výpočtu, toto pozdeji decomissionovat
#lookback has to be even
if lookback_offset % 2 != 0:
lookback_offset += 1
#TBD pripdadne /2
if len(state.bars.close) > (slope_lookback + lookback_offset):
#test prumer nejvyssi a nejnizsi hodnoty
# if name == "slope":
#levy bod bude vzdy vzdaleny o slope_lookback
#ten bude prumerem hodnot lookback_offset a to tak ze polovina offsetu z kazde strany
array_od = slope_lookback + int(lookback_offset/2)
array_do = slope_lookback - int(lookback_offset/2)
#lookbackprice_array = state.bars.vwap[-array_od:-array_do]
#lookbackprice = round(sum(lookbackprice_array)/lookback_offset,3)
#jako optimalizace pouzijeme NUMPY
lookbackprice = np.mean(state.bars.vwap[-array_od:-array_do])
# Round the lookback price to 3 decimal places
lookbackprice = round(lookbackprice, 3)
#lookbackprice = round((min(lookbackprice_array)+max(lookbackprice_array))/2,3)
# else:
# #puvodni lookback a od te doby dozadu offset
# array_od = slope_lookback + lookback_offset
# array_do = slope_lookback
# lookbackprice_array = state.bars.vwap[-array_od:-array_do]
# #obycejný prumer hodnot
# lookbackprice = round(sum(lookbackprice_array)/lookback_offset,3)
lookbacktime = state.bars.time[-slope_lookback]
else:
#kdyz neni dostatek hodnot, pouzivame jako levy bod open hodnotu close[0]
#lookbackprice = state.bars.vwap[0]
#dalsi vyarianta-- lookback je pole z toho všeho co mame
#lookbackprice = Average(state.bars.vwap)
#pokud neni dostatek, bereme vzdy prvni petinu z dostupnych barů
# a z ní uděláme průměr
cnt = len(state.bars.close)
if cnt>5:
sliced_to = int(cnt/5)
lookbackprice = np.mean(state.bars.vwap[:sliced_to])
#lookbackprice= Average(state.bars.vwap[:sliced_to])
lookbacktime = state.bars.time[int(sliced_to/2)]
else:
lookbackprice = np.mean(state.bars.vwap)
#lookbackprice = Average(state.bars.vwap)
lookbacktime = state.bars.time[0]
state.ilog(lvl=1,e=f"IND {name} slope - not enough data bereme left bod open", slope_lookback=slope_lookback, lookbackprice=lookbackprice)
#výpočet úhlu - a jeho normalizace
slope = ((state.bars.close[-1] - lookbackprice)/lookbackprice)*100
slope = round(slope, 4)
state.indicators[name][-1]=slope
#angle je ze slope, ale pojmenovavame ho podle MA
state.statinds[name] = dict(time=state.bars.time[-1], price=state.bars.close[-1], lookbacktime=lookbacktime, lookbackprice=lookbackprice, minimum_slope=minimum_slope, maximum_slope=maximum_slope)
#slope MA vyrovna vykyvy ve slope
slope_MA_length = safe_get(options, 'MA_length', None)
slopeMA = None
last_slopesMA = None
#pokud je nastavena MA_length tak vytvarime i MAcko dane delky na tento slope
if slope_MA_length is not None:
source = state.indicators[name][-slope_MA_length:]
slopeMAseries = ema(source, slope_MA_length) #state.bars.vwap
slopeMA = round(slopeMAseries[-1],4)
state.indicators[name+"MA"][-1]=slopeMA
last_slopesMA = state.indicators[name+"MA"][-10:]
lb_priceline_string = "from "+lookback_priceline if lookback_priceline is not None else ""
state.ilog(lvl=1,e=f"IND {name} {lb_priceline_string} {slope=} {slopeMA=}", msg=f"{lookbackprice=} {lookbacktime=}", lookback_priceline=lookback_priceline, lookbackprice=lookbackprice, lookbacktime=lookbacktime, slope_lookback=slope_lookback, lookbackoffset=lookback_offset, minimum_slope=minimum_slope, last_slopes=state.indicators[name][-10:], last_slopesMA=last_slopesMA)
#dale pracujeme s timto MAckovanym slope
#slope = slopeMA
except Exception as e:
print(f"Exception in {name} slope Indicator section", str(e))
state.ilog(lvl=1,e=f"EXCEPTION in {name}", msg="Exception in slope Indicator section" + str(e) + format_exc())

124
v2realbot/strategyblocks/indicators/slopeLP.py Normal file
View File

@ -0,0 +1,124 @@
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.strategy.base import StrategyState
from v2realbot.indicators.indicators import ema, natr, roc
from v2realbot.indicators.oscillators import rsi
from traceback import format_exc
#SLOPE LP
def populate_dynamic_slopeLP_indicator(data, state: StrategyState, name):
ind_type = "slopeLP"
options = safe_get(state.vars.indicators, name, None)
if options is None:
state.ilog(lvl=1,e=f"No options for {name} in stratvars")
return
if safe_get(options, "type", False) is False or safe_get(options, "type", False) != ind_type:
state.ilog(lvl=1,e="Type error")
return
#poustet kazdy tick nebo jenom na confirmed baru (on_confirmed_only = true)
on_confirmed_only = safe_get(options, 'on_confirmed_only', False)
#pocet baru po kterých se levy bod z BUY prepne opet na standadni vypocet (prumer)
#kdyz se dlouho neprodává a cena nejde dolu, tak aby se nezastavilo nakupovani
back_to_standard_after = int(safe_get(options, 'back_to_standard_after', 0))
#slopeLP INDIKATOR
#levy bod je nejdrive standardne automaticky vypočtený podle hodnoty lookbacku (např. -8, offset 4)
#při nákupu se BUY POINT se stává levým bodem (až do doby kdy není lookbackprice nižší, pak pokračuje lookbackprice)
#při prodeji se SELL POINT se stává novým levým bodem (až do doby kdy není lookbackprice vyšší, pak pokračuje lookbackprice)
#zatím implementovat prvni část (mimo části ..až do doby) - tu pak dodelat podle vysledku, pripadne ji neimplementovat vubec a misto toho
#udelat slope RESET pri dosazeni urciteho pozitivniho nebo negativni slopu
#zkusime nejdriv: levy bod automat, po nakupu je levy bod cena nakupu
#VYSTUPY: state.indicators[name],
# state.indicators[nameMA]
# statický indikátor (angle) - stejneho jmena pro vizualizaci uhlu
if on_confirmed_only is False or (on_confirmed_only is True and data['confirmed']==1):
try:
#slow_slope = 99
slope_lookback = safe_get(options, 'slope_lookback', 100)
minimum_slope = safe_get(options, 'minimum_slope', 25)
maximum_slope = safe_get(options, "maximum_slope",0.9)
lookback_offset = safe_get(options, 'lookback_offset', 25)
#typ leveho bodu [lastbuy - cena posledniho nakupu, baropen - cena otevreni baru]
leftpoint = safe_get(options, 'leftpoint', "lastbuy")
#lookback has to be even
if lookback_offset % 2 != 0:
lookback_offset += 1
if leftpoint == "lastbuy":
if len(state.bars.close) > (slope_lookback + lookback_offset):
#test prumer nejvyssi a nejnizsi hodnoty
# if name == "slope":
#levy bod bude vzdy vzdaleny o slope_lookback
#ten bude prumerem hodnot lookback_offset a to tak ze polovina offsetu z kazde strany
array_od = slope_lookback + int(lookback_offset/2)
array_do = slope_lookback - int(lookback_offset/2)
lookbackprice_array = state.bars.vwap[-array_od:-array_do]
#cas nastavujeme vzdy podle nastaveni (zatim)
lookbacktime = state.bars.time[-slope_lookback]
#pokud mame aktivni pozice, nastavime lookbackprice a time podle posledniho tradu
#pokud se ale dlouho nenakupuje (uplynulo od posledniho nakupu vic nez back_to_standard_after baru), tak se vracime k prumeru
if state.avgp > 0 and state.bars.index[-1] < int(state.vars.last_buy_index)+back_to_standard_after:
lb_index = -1 - (state.bars.index[-1] - int(state.vars.last_buy_index))
lookbackprice = state.bars.vwap[lb_index]
state.ilog(lvl=0,e=f"IND {name} slope {leftpoint}- LEFT POINT OVERRIDE bereme ajko cenu lastbuy {lookbackprice=} {lookbacktime=} {lb_index=}")
else:
#dame na porovnani jen prumer
lookbackprice = round(sum(lookbackprice_array)/lookback_offset,3)
#lookbackprice = round((min(lookbackprice_array)+max(lookbackprice_array))/2,3)
# else:
# #puvodni lookback a od te doby dozadu offset
# array_od = slope_lookback + lookback_offset
# array_do = slope_lookback
# lookbackprice_array = state.bars.vwap[-array_od:-array_do]
# #obycejný prumer hodnot
# lookbackprice = round(sum(lookbackprice_array)/lookback_offset,3)
lookbacktime = state.bars.time[-slope_lookback]
state.ilog(lvl=0,e=f"IND {name} slope {leftpoint} - LEFT POINT STANDARD {lookbackprice=} {lookbacktime=}")
else:
#kdyz neni dostatek hodnot, pouzivame jako levy bod open hodnotu close[0]
lookbackprice = state.bars.close[0]
lookbacktime = state.bars.time[0]
state.ilog(lvl=0,e=f"IND {name} slope - not enough data bereme left bod open", slope_lookback=slope_lookback)
elif leftpoint == "baropen":
lookbackprice = state.bars.open[-1]
lookbacktime = state.bars.time[-1]
state.ilog(lvl=0,e=f"IND {name} slope {leftpoint}- bereme cenu bar OPENu ", lookbackprice=lookbackprice, lookbacktime=lookbacktime)
else:
state.ilog(lvl=0,e=f"IND {name} UNKNOW LEFT POINT TYPE {leftpoint=}")
#výpočet úhlu - a jeho normalizace
slope = ((state.bars.close[-1] - lookbackprice)/lookbackprice)*100
slope = round(slope, 4)
state.indicators[name][-1]=slope
#angle ze slope
state.statinds[name] = dict(time=state.bars.updated[-1], price=state.bars.close[-1], lookbacktime=lookbacktime, lookbackprice=lookbackprice, minimum_slope=minimum_slope, maximum_slope=maximum_slope)
#slope MA vyrovna vykyvy ve slope
slope_MA_length = safe_get(options, 'MA_length', None)
slopeMA = None
last_slopesMA = None
#pokud je nastavena MA_length tak vytvarime i MAcko dane delky na tento slope
if slope_MA_length is not None:
source = state.indicators[name][-slope_MA_length:]
slopeMAseries = ema(source, slope_MA_length) #state.bars.vwap
slopeMA = round(slopeMAseries[-1],5)
state.indicators[name+"MA"][-1]=slopeMA
last_slopesMA = state.indicators[name+"MA"][-10:]
state.ilog(lvl=0,e=f"{name=} {slope=} {slopeMA=}", msg=f"{lookbackprice=}", lookbackoffset=lookback_offset, minimum_slope=minimum_slope, last_slopes=state.indicators[name][-10:], last_slopesMA=last_slopesMA)
#dale pracujeme s timto MAckovanym slope
#slope = slopeMA
except Exception as e:
print(f"Exception in {name} slope Indicator section", str(e))
state.ilog(lvl=1,e=f"EXCEPTION in {name}", msg="Exception in slope Indicator section" + str(e) + format_exc())

84
v2realbot/strategyblocks/inits/init_directives.py Normal file
View File

@ -0,0 +1,84 @@
from v2realbot.strategy.base import StrategyState
from v2realbot.strategy.StrategyOrderLimitVykladaciNormalizedMYSELL import StrategyOrderLimitVykladaciNormalizedMYSELL
from v2realbot.enums.enums import RecordType, StartBarAlign, Mode, Account, Followup
from v2realbot.common.PrescribedTradeModel import Trade, TradeDirection, TradeStatus
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.utils.directive_utils import get_conditions_from_configuration
from v2realbot.ml.mlutils import load_model
from v2realbot.common.model import SLHistory
from v2realbot.config import KW
from uuid import uuid4
from datetime import datetime
#import random
import json
import numpy as np
#from icecream import install, ic
from rich import print as printanyway
from threading import Event
import os
from traceback import format_exc
def intialize_directive_conditions(state):
#inciializace pro akce: short, long, dont_short, dont_long, activate
state.vars.conditions = {}
#KEYWORDS_if_CONDITION = value
# např. go_short_if_below = 10
#possible KEYWORDS in directive: (AND/OR) support
# go_DIRECTION(go_long_if, go_short_if)
# dont_go_DIRECTION (dont_long_if, dont_short_if)
# exit_DIRECTION (exit_long_if, exit_short_if)
# activate (activate_if)
#possible CONDITIONs:
# below, above, falling, rising, crossed_up, crossed_down
#Tyto mohou byt bud v sekci conditions a nebo v samostatne sekci common
#pro kazdou sekci "conditions" v signals
#si vytvorime podminkove dictionary pro kazdou akci
#projdeme vsechny singaly
#nejprve genereujeme ze SIGNALu
for signalname, signalsettings in state.vars.signals.items():
if "conditions" in signalsettings:
section = signalsettings["conditions"]
#directivy non direction related
state.vars.conditions.setdefault(KW.activate,{})[signalname] = get_conditions_from_configuration(action=KW.activate+"_if", section=section)
#direktivy direction related
for smer in TradeDirection:
#IDEA navrhy condition dictionary - ty v signal sekci
# state.vars.conditions["nazev_evaluacni_sekce"]["nazevsignalu_smer"] = #sada podminek
#signal related
# state.vars.conditions["activate"]["trendfollow"] = #sada podminek
# state.vars.conditions["dont_go"]["trendfollow"]["long"] = #sada podminek
# state.vars.conditions["go"]["trendfollow"]["short"] = #sada podminek
# state.vars.conditions["exit"]["trendfollow"]["long"] = #sada podminek
#common
# state.vars.conditions["exit"]["common"]["long"] = #sada podminek
# state.vars.conditions["exit"]["common"]["long"] = #sada podminek
state.vars.conditions.setdefault(KW.dont_go,{}).setdefault(signalname,{})[smer] = get_conditions_from_configuration(action=KW.dont_go+"_" + smer +"_if", section=section)
state.vars.conditions.setdefault(KW.dont_exit,{}).setdefault(signalname,{})[smer] = get_conditions_from_configuration(action=KW.dont_exit+"_" + smer +"_if", section=section)
state.vars.conditions.setdefault(KW.go,{}).setdefault(signalname,{})[smer] = get_conditions_from_configuration(action=KW.go+"_" + smer +"_if", section=section)
state.vars.conditions.setdefault(KW.exit,{}).setdefault(signalname,{})[smer] = get_conditions_from_configuration(action=KW.exit+"_" + smer +"_if", section=section)
state.vars.conditions.setdefault(KW.reverse,{}).setdefault(signalname,{})[smer] = get_conditions_from_configuration(action=KW.reverse+"_" + smer +"_if", section=section)
state.vars.conditions.setdefault(KW.exitadd,{}).setdefault(signalname,{})[smer] = get_conditions_from_configuration(action=KW.exitadd+"_" + smer +"_if", section=section)
state.vars.conditions.setdefault(KW.slreverseonly,{}).setdefault(signalname,{})[smer] = get_conditions_from_configuration(action=KW.slreverseonly+"_" + smer +"_if", section=section)
# 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")
#POTOM generujeme z obecnych sekci, napr. EXIT.EXIT_CONDITIONS, kde je fallback pro signal exity
section = state.vars.exit["conditions"]
for smer in TradeDirection:
state.vars.conditions.setdefault(KW.exit,{}).setdefault("common",{})[smer] = get_conditions_from_configuration(action=KW.exit+"_" + smer +"_if", section=section)
state.vars.conditions.setdefault(KW.dont_exit,{}).setdefault("common",{})[smer] = get_conditions_from_configuration(action=KW.dont_exit+"_" + smer +"_if", section=section)
state.vars.conditions.setdefault(KW.reverse,{}).setdefault("common",{})[smer] = get_conditions_from_configuration(action=KW.reverse+"_" + smer +"_if", section=section)
state.vars.conditions.setdefault(KW.exitadd,{}).setdefault("common",{})[smer] = get_conditions_from_configuration(action=KW.exitadd+"_" + smer +"_if", section=section)
state.vars.conditions.setdefault(KW.slreverseonly,{}).setdefault("common",{})[smer] = get_conditions_from_configuration(action=KW.slreverseonly+"_" + smer +"_if", section=section)

60
v2realbot/strategyblocks/inits/init_indicators.py Normal file
View File

@ -0,0 +1,60 @@
from v2realbot.strategy.base import StrategyState
from v2realbot.strategy.StrategyOrderLimitVykladaciNormalizedMYSELL import StrategyOrderLimitVykladaciNormalizedMYSELL
from v2realbot.enums.enums import RecordType, StartBarAlign, Mode, Account, Followup
from v2realbot.common.PrescribedTradeModel import Trade, TradeDirection, TradeStatus
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.utils.directive_utils import get_conditions_from_configuration
from v2realbot.ml.mlutils import load_model
from v2realbot.common.model import SLHistory
from v2realbot.config import KW
from uuid import uuid4
from datetime import datetime
#import random
import json
import numpy as np
#from icecream import install, ic
from rich import print as printanyway
from threading import Event
import os
from traceback import format_exc
def initialize_dynamic_indicators(state):
#pro vsechny indikatory, ktere maji ve svych stratvars TYPE inicializujeme
dict_copy = state.vars.indicators.copy()
for indname, indsettings in dict_copy.items():
for option,value in list(indsettings.items()):
#inicializujeme nejenom typizovane
#if option == "type":
state.indicators[indname] = []
#pokud ma MA_length incializujeme i MA variantu
if safe_get(indsettings, 'MA_length', False):
state.indicators[indname+"MA"] = []
#specifika pro slope
if option == "type":
if value == "slope":
#inicializujeme statinds (pro uhel na FE)
state.statinds[indname] = dict(minimum_slope=safe_get(indsettings, 'minimum_slope', -1), maximum_slope=safe_get(indsettings, 'maximum_slope', 1))
if value == "custom":
#pro typ custom inicializujeme promenne
state.vars.indicators[indname]["last_run_time"] = None
state.vars.indicators[indname]["last_run_index"] = None
if option == "subtype":
if value == "model":
active = safe_get(indsettings, 'active', True)
if active is False:
continue
#load the model
modelname = safe_get(indsettings["cp"], 'name', None)
modelversion = safe_get(indsettings["cp"], 'version', "1")
if modelname is not None:
state.vars.loaded_models[modelname] = load_model(modelname, modelversion)
if state.vars.loaded_models[modelname] is not None:
printanyway(f"model {modelname} loaded")
else:
printanyway(f"ERROR model {modelname} NOT loaded")
#pro conditional indikatory projedeme podminky [conditions] a pro kazdou pripravime (cond_dict)
if value == "conditional":
conditions = state.vars.indicators[indname]["cp"]["conditions"]
for condname,condsettings in conditions.items():
state.vars.indicators[indname]["cp"]["conditions"][condname]["cond_dict"] = get_conditions_from_configuration(action=KW.change_val+"_if", section=condsettings)
printanyway(f'creating workdict for {condname} value {state.vars.indicators[indname]["cp"]["conditions"][condname]["cond_dict"]}')

163
v2realbot/strategyblocks/newtrade/conditions.py Normal file
View File

@ -0,0 +1,163 @@
from v2realbot.strategy.base import StrategyState
from v2realbot.strategy.StrategyOrderLimitVykladaciNormalizedMYSELL import StrategyOrderLimitVykladaciNormalizedMYSELL
from v2realbot.enums.enums import RecordType, StartBarAlign, Mode, Account, Followup
from v2realbot.common.PrescribedTradeModel import Trade, TradeDirection, TradeStatus
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get, is_still, is_window_open, eval_cond_dict, crossed_down, crossed_up, crossed, is_pivot, json_serial, pct_diff, create_new_bars, slice_dict_lists
from v2realbot.utils.directive_utils import get_conditions_from_configuration
from v2realbot.ml.mlutils import load_model
from v2realbot.common.model import SLHistory
from v2realbot.config import KW
from uuid import uuid4
from datetime import datetime
#import random
import json
import numpy as np
#from icecream import install, ic
from rich import print as printanyway
from threading import Event
import os
from traceback import format_exc
from v2realbot.strategyblocks.indicators.indicators_hub import populate_all_indicators
from v2realbot.strategyblocks.indicators.helpers import evaluate_directive_conditions
#preconditions and conditions of LONG/SHORT SIGNAL
def go_conditions_met(state, data, signalname: str, direction: TradeDirection):
if direction == TradeDirection.LONG:
smer = "long"
else:
smer = "short"
#preconditiony dle smer
#SPECIFICKE DONT BUYS - direktivy zacinajici dont_buy
#dont_buy_below = value nebo nazev indikatoru
#dont_buy_above = value nebo hazev indikatoru
#TESTUJEME SPECIFICKY DONT_GO -
#u techto ma smysl pouze OR
cond_dict = state.vars.conditions[KW.dont_go][signalname][smer]
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "OR")
state.ilog(lvl=1,e=f"SPECIFIC PRECOND {smer} {result}", **conditions_met, cond_dict=cond_dict)
if result:
return False
# #OR neprosly testujeme AND
# result, conditions_met = evaluate_directive_conditions(cond_dict, "AND")
# state.ilog(lvl=0,e=f"EXIT CONDITIONS of activeTrade {smer} =AND= {result}", **conditions_met, cond_dict=cond_dict)
# if result:
# return True
#tyto timto nahrazeny - dat do konfigurace (dont_short_when, dont_long_when)
#dont_buy_when['rsi_too_high'] = state.indicators.RSI14[-1] > safe_get(state.vars, "rsi_dont_buy_above",50)
#dont_buy_when['slope_too_low'] = slope_too_low()
#dont_buy_when['slope_too_high'] = slope_too_high()
#dont_buy_when['rsi_is_zero'] = (state.indicators.RSI14[-1] == 0)
#dont_buy_when['reverse_position_waiting_amount_not_0'] = (state.vars.reverse_position_waiting_amount != 0)
#u indikatoru muzoun byt tyto directivy pro generovani signaliu long/short
# long_if_crossed_down - kdyz prekrocil dolu, VALUE: hodnota nebo nazev indikatoru
# long_if_crossed_up - kdyz prekrocil nahoru, VALUE: hodnota nebo nazev indikatoru
# long_if_crossed - kdyz krosne obema smery, VALUE: hodnota nebo nazev indikatoru
# long_if_falling - kdyz je klesajici po N, VALUE: hodnota
# long_if_rising - kdyz je rostouci po N, VALUE: hodnota
# long_if_below - kdyz je pod prahem, VALUE: hodnota nebo nazev indikatoru
# long_if_above - kdyz je nad prahem, VALUE: hodnota nebo nazev indikatoru
# long_if_pivot_a - kdyz je pivot A. VALUE: delka nohou
# long_if_pivot_v - kdyz je pivot V. VALUE: delka nohou
# direktivy se mohou nachazet v podsekci AND nebo OR - daneho indikatoru (nebo na volno, pak = OR)
# OR - staci kdyz plati jedna takova podminka a buysignal je aktivni
# AND - musi platit vsechny podminky ze vsech indikatoru, aby byl buysignal aktivni
#populate work dict - muze byt i jen jednou v INIT nebo 1x za cas
#dict oindexovane podminkou (OR/AND) obsahuje vsechny buy_if direktivy v tuplu (nazevind,direktiva,hodnota
# {'AND': [('nazev indikatoru', 'nazev direktivy', 'hodnotadirektivy')], 'OR': []}
#work_dict_signal_if = get_work_dict_with_directive(starts_with=signalname+"_"+smer+"_if")
#TESTUJEME GO SIGNAL
cond_dict = state.vars.conditions[KW.go][signalname][smer]
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "OR")
state.ilog(lvl=1,e=f"EVAL GO SIGNAL {smer} =OR= {result}", **conditions_met, cond_dict=cond_dict)
if result:
return True
#OR neprosly testujeme AND
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "AND")
state.ilog(lvl=1,e=f"EVAL GO SIGNAL {smer} =AND= {result}", **conditions_met, cond_dict=cond_dict)
if result:
return True
return False
#obecne precondition preds vstupem - platne jak pro condition based tak pro plugin
def common_go_preconditions_check(state, data, signalname: str, options: dict):
#ZAKLADNI KONTROLY ATRIBUTU s fallbackem na obecné
#check working windows (open - close, in minutes from the start of marker)
window_open = safe_get(options, "window_open",safe_get(state.vars, "window_open",0))
window_close = safe_get(options, "window_close",safe_get(state.vars, "window_close",390))
if is_window_open(datetime.fromtimestamp(data['updated']).astimezone(zoneNY), window_open, window_close) is False:
state.ilog(lvl=1,e=f"SIGNAL {signalname} - WINDOW CLOSED", msg=f"{window_open=} {window_close=} ")
return False
min_bar_index = safe_get(options, "min_bar_index",safe_get(state.vars, "min_bar_index",0))
if int(data["index"]) < int(min_bar_index):
state.ilog(lvl=1,e=f"MIN BAR INDEX {min_bar_index} waiting - TOO SOON", currindex=data["index"])
return False
next_signal_offset = safe_get(options, "next_signal_offset_from_last_exit",safe_get(state.vars, "next_signal_offset_from_last_exit",0))
if state.vars.last_exit_index is not None:
index_to_compare = int(state.vars.last_exit_index)+int(next_signal_offset)
if index_to_compare > int(data["index"]):
state.ilog(lvl=1,e=f"NEXT SIGNAL OFFSET from EXIT {next_signal_offset} waiting - TOO SOON", currindex=data["index"], index_to_compare=index_to_compare, last_exit_index=state.vars.last_exit_index)
return False
# if is_open_rush(datetime.fromtimestamp(data['updated']).astimezone(zoneNY), open_rush) or is_close_rush(datetime.fromtimestamp(data['updated']).astimezone(zoneNY), close_rush):
# state.ilog(lvl=0,e=f"SIGNAL {signalname} - WINDOW CLOSED", msg=f"{open_rush=} {close_rush=} ")
# return False
#natvrdo nebo na podminku
activated = safe_get(options, "activated", True)
#check activation
if activated is False:
state.ilog(lvl=1,e=f"{signalname} not ACTIVATED")
cond_dict = state.vars.conditions[KW.activate][signalname]
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "OR")
state.ilog(lvl=1,e=f"EVAL ACTIVATION CONDITION =OR= {result}", **conditions_met, cond_dict=cond_dict)
if result is False:
#OR neprosly testujeme AND
result, conditions_met = evaluate_directive_conditions(state, cond_dict, "AND")
state.ilog(lvl=1,e=f"EVAL ACTIVATION CONDITION =AND= {result}", **conditions_met, cond_dict=cond_dict)
if result is False:
state.ilog(lvl=1,e=f"not ACTIVATED")
return False
else:
state.ilog(lvl=1,e=f"{signalname} JUST ACTIVATED")
state.vars.signals[signalname]["activated"] = True
# OBECNE PRECONDITIONS - typu dont_do_when
precond_check = dict(AND=dict(), OR=dict())
# #OBECNE DONT BUYS
if safe_get(options, "signal_only_on_confirmed",safe_get(state.vars, "signal_only_on_confirmed",True)):
precond_check['bar_not_confirmed'] = (data['confirmed'] == 0)
# #od posledniho vylozeni musi ubehnout N baru
# dont_buy_when['last_buy_offset_too_soon'] = data['index'] < (int(state.vars.lastbuyindex) + int(safe_get(state.vars, "lastbuy_offset",3)))
# dont_buy_when['blockbuy_active'] = (state.vars.blockbuy == 1)
# dont_buy_when['jevylozeno_active'] = (state.vars.jevylozeno == 1)
#obecne open_rush platne pro vsechny
#precond_check['on_confirmed_only'] = safe_get(options, 'on_confirmed_only', False) - chybi realizace podminky, pripadne dodelat na short_on_confirmed
# #testing preconditions
result, cond_met = eval_cond_dict(precond_check)
if result:
state.ilog(lvl=1,e=f"PRECOND GENERAL not met {cond_met}", message=cond_met, precond_check=precond_check)
return False
state.ilog(lvl=1,e=f"{signalname} ALL PRECOND MET")
return True

2
v2realbot/strategyblocks/newtrade/plugins/asr.py Normal file
View File

@ -0,0 +1,2 @@
#ASR signal plugin
#WIP

78
v2realbot/strategyblocks/newtrade/prescribedtrades.py Normal file
View File

@ -0,0 +1,78 @@
from v2realbot.strategy.base import StrategyState
from v2realbot.common.PrescribedTradeModel import TradeDirection, TradeStatus
from v2realbot.utils.utils import zoneNY, json_serial
from datetime import datetime
#import random
import json
from v2realbot.strategyblocks.activetrade.helpers import insert_SL_history, get_default_sl_value, normalize_tick
def execute_prescribed_trades(state: StrategyState, data):
##evaluate prescribed trade, prvni eligible presuneme do activeTrade, zmenime stav and vytvorime objednavky
if state.vars.activeTrade is not None or len(state.vars.prescribedTrades) == 0:
return
#evaluate long (price/market)
state.ilog(lvl=1,e="evaluating prescr trades", trades=json.loads(json.dumps(state.vars.prescribedTrades, default=json_serial)))
for trade in state.vars.prescribedTrades:
if trade.status == TradeStatus.READY and trade.direction == TradeDirection.LONG and (trade.entry_price is None or trade.entry_price >= data['close']):
trade.status = TradeStatus.ACTIVATED
trade.last_update = datetime.fromtimestamp(state.time).astimezone(zoneNY)
state.ilog(lvl=1,e=f"evaluated LONG", trade=json.loads(json.dumps(trade, default=json_serial)), prescrTrades=json.loads(json.dumps(state.vars.prescribedTrades, default=json_serial)))
state.vars.activeTrade = trade
state.vars.last_buy_index = data["index"]
state.vars.last_in_index = data["index"]
break
#evaluate shorts
if not state.vars.activeTrade:
for trade in state.vars.prescribedTrades:
if trade.status == TradeStatus.READY and trade.direction == TradeDirection.SHORT and (trade.entry_price is None or trade.entry_price <= data['close']):
state.ilog(lvl=1,e=f"evaluaed SHORT", trade=json.loads(json.dumps(trade, default=json_serial)), prescTrades=json.loads(json.dumps(state.vars.prescribedTrades, default=json_serial)))
trade.status = TradeStatus.ACTIVATED
trade.last_update = datetime.fromtimestamp(state.time).astimezone(zoneNY)
state.vars.activeTrade = trade
state.vars.last_buy_index = data["index"]
state.vars.last_in_index = data["index"]
break
#odeslani ORDER + NASTAVENI STOPLOSS (zatim hardcoded)
if state.vars.activeTrade:
if state.vars.activeTrade.direction == TradeDirection.LONG:
state.ilog(lvl=1,e="odesilame LONG ORDER", trade=json.loads(json.dumps(state.vars.activeTrade, default=json_serial)))
if state.vars.activeTrade.size is not None:
size = state.vars.activeTrade.size
else:
size = state.vars.chunk
res = state.buy(size=size)
if isinstance(res, int) and res < 0:
raise Exception(f"error in required operation LONG {res}")
#nastaveni SL az do notifikace, kdy je známá
#pokud neni nastaveno SL v prescribe, tak nastavuji default dle stratvars
if state.vars.activeTrade.stoploss_value is None:
sl_defvalue = get_default_sl_value(state, direction=state.vars.activeTrade.direction)
#normalizuji dle aktualni ceny
sl_defvalue_normalized = normalize_tick(state, data,sl_defvalue)
state.vars.activeTrade.stoploss_value = float(data['close']) - sl_defvalue_normalized
insert_SL_history(state)
state.ilog(lvl=1,e=f"Nastaveno SL na {sl_defvalue}, priced normalized: {sl_defvalue_normalized} price: {state.vars.activeTrade.stoploss_value }")
state.vars.pending = state.vars.activeTrade.id
elif state.vars.activeTrade.direction == TradeDirection.SHORT:
state.ilog(lvl=1,e="odesilame SHORT ORDER",trade=json.loads(json.dumps(state.vars.activeTrade, default=json_serial)))
if state.vars.activeTrade.size is not None:
size = state.vars.activeTrade.size
else:
size = state.vars.chunk
res = state.sell(size=size)
if isinstance(res, int) and res < 0:
raise Exception(f"error in required operation SHORT {res}")
#pokud neni nastaveno SL v prescribe, tak nastavuji default dle stratvars
if state.vars.activeTrade.stoploss_value is None:
sl_defvalue = get_default_sl_value(state, direction=state.vars.activeTrade.direction)
#normalizuji dle aktualni ceny
sl_defvalue_normalized = normalize_tick(state, data, sl_defvalue)
state.vars.activeTrade.stoploss_value = float(data['close']) + sl_defvalue_normalized
insert_SL_history(state)
state.ilog(lvl=1,e=f"Nastaveno SL na {sl_defvalue}, priced normalized: {sl_defvalue_normalized} price: {state.vars.activeTrade.stoploss_value }")
state.vars.pending = state.vars.activeTrade.id
else:
state.ilog(lvl=1,e="unknow direction")
state.vars.activeTrade = None

92
v2realbot/strategyblocks/newtrade/signals.py Normal file
View File

@ -0,0 +1,92 @@
from v2realbot.strategy.base import StrategyState
from v2realbot.common.PrescribedTradeModel import Trade, TradeDirection, TradeStatus
from v2realbot.utils.utils import isrising, isfalling,zoneNY, price2dec, print, safe_get
from v2realbot.config import KW
from uuid import uuid4
from datetime import datetime
from rich import print as printanyway
from traceback import format_exc
from v2realbot.strategyblocks.newtrade.conditions import go_conditions_met, common_go_preconditions_check
def signal_search(state: StrategyState, data):
# SIGNAL sekce ve stratvars obsahuje signaly: Ty se skladaji z obecnych parametru a podsekce podminek.
# Obecne parametry mohou overridnout root parametry nebo dalsi upresneni(napr. plugin). Podsekce CONDITIONS,obsahuji podminky vstup a vystupu
# OBECNE:
# [stratvars.signals.trend2]
# signal_only_on_confirmed = true
# open_rush = 2
# close_rush = 6000
# short_enabled = false
# long_enabled = false
# activated = true
# profit = 0.2
# max_profit = 0.4
# PODMINKY:
# [stratvars.signals.trend2.conditions]
# slope20.AND.in_long_if_above = 0.23
# slope10.AND.in_long_if_rising = 5
# slope10.out_long_if_crossed_down = -0.1
# slope10.in_short_if_crossed_down = -0.1
# slope10.out_short_if_above = 0
# ema.AND.short_if_below = 28
for signalname, signalsettings in state.vars.signals.items():
execute_signal_generator(state, data, signalname)
# #vysledek je vložení Trade Prescription a to bud s cenou nebo immediate
# pokud je s cenou ceka se na cenu, pokud immmediate tak se hned provede
# to vse za predpokladu, ze neni aktivni trade
def execute_signal_generator(state, data, name):
state.ilog(lvl=0,e=f"SIGNAL SEARCH for {name}", cond_go=state.vars.conditions[KW.go][name], cond_dontgo=state.vars.conditions[KW.dont_go][name], cond_activate=state.vars.conditions[KW.activate][name] )
options = safe_get(state.vars.signals, name, None)
if options is None:
state.ilog(lvl=1,e="No options for {name} in stratvars")
return
if common_go_preconditions_check(state, data, signalname=name, options=options) is False:
return
# signal_plugin = "reverzni"
# signal_plugin_run_once_at_index = 3
#pokud existuje plugin, tak pro signal search volame plugin a ignorujeme conditiony
signal_plugin = safe_get(options, 'plugin', None)
signal_plugin_run_once_at_index = safe_get(options, 'signal_plugin_run_once_at_index', 3)
#pokud je plugin True, spusti se kod
if signal_plugin is not None and signal_plugin_run_once_at_index==data["index"]:
try:
custom_function = eval(signal_plugin)
custom_function()
except NameError:
state.ilog(lvl=1,e="Custom plugin {signal_plugin} not found")
else:
short_enabled = safe_get(options, "short_enabled",safe_get(state.vars, "short_enabled",True))
long_enabled = safe_get(options, "long_enabled",safe_get(state.vars, "long_enabled",True))
#common signals based on 1) configured signals in stratvars
#toto umoznuje jednoduchy prescribed trade bez ceny
if short_enabled is False:
state.ilog(lvl=1,e=f"{name} SHORT DISABLED")
if long_enabled is False:
state.ilog(lvl=1,e=f"{name} LONG DISABLED")
if long_enabled and go_conditions_met(state, data,signalname=name, direction=TradeDirection.LONG):
state.vars.prescribedTrades.append(Trade(
id=uuid4(),
last_update=datetime.fromtimestamp(state.time).astimezone(zoneNY),
status=TradeStatus.READY,
generated_by=name,
direction=TradeDirection.LONG,
entry_price=None,
stoploss_value = None))
elif short_enabled and go_conditions_met(state, data, signalname=name, direction=TradeDirection.SHORT):
state.vars.prescribedTrades.append(Trade(
id=uuid4(),
last_update=datetime.fromtimestamp(state.time).astimezone(zoneNY),
status=TradeStatus.READY,
generated_by=name,
direction=TradeDirection.SHORT,
entry_price=None,
stoploss_value = None))
else:
state.ilog(lvl=0,e=f"{name} NO SIGNAL")