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

other changes

Browse Source
This commit is contained in:
David Brazda
2023-11-15 09:02:15 +01:00
parent aead08a2c9
commit dc4c10a4a3
15 changed files with 380 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
v2realbot/strategyblocks/activetrade/helpers.py
View File

@ -161,23 +161,32 @@ def get_profit_target_price(state, data, direction: TradeDirection):
#mame v direktivve ticky
if isinstance(def_profit, (float, int)):
normalized_def_profit = normalize_tick(state, data, float(def_profit))
state.ilog(lvl=0,e=f"PROFIT {def_profit=} {normalized_def_profit=}")
base_price = state.avgp if state.avgp != 0 else data["close"]
to_return = price2dec(float(base_price)+normalized_def_profit,3) if direction == TradeDirection.LONG else price2dec(float(base_price)-normalized_def_profit,3)
to_return = get_normalized_profitprice_from_tick(state, data, def_profit, direction)
#mame v direktive indikator
elif isinstance(def_profit, str):
to_return = float(value_or_indicator(state, def_profit))
if direction == TradeDirection.LONG and to_return < data['close'] or direction == TradeDirection.SHORT and to_return > data['close']:
state.ilog(lvl=1,e=f"SPATNA HODOTA DOTAZENEHO PROFITU z ind {def_profit} {to_return=} {smer} {data['close']}")
raise Exception(f"SPATNA HODOTA DOTAZENEHO PROFITU z ind{def_profit} {to_return=} {smer} {data['close']}")
state.ilog(lvl=1,e=f"DOTAZENY PROFIT z indikatoru {def_profit} {to_return=}")
#min profit (ochrana extremnich hodnot indikatoru)
directive_name = 'profit_min_ind_tick_value'
profit_min_ind_tick_value = get_override_for_active_trade(state, directive_name=directive_name, default_value=def_profit_both_directions)
profit_min_ind_price_value = get_normalized_profitprice_from_tick(state, data, profit_min_ind_tick_value, direction)
#ochrana pri nastaveni profitu prilis nizko
if direction == TradeDirection.LONG and to_return < profit_min_ind_price_value or direction == TradeDirection.SHORT and to_return > profit_min_ind_price_value:
state.ilog(lvl=1,e=f"SPATNA HODOTA DOTAZENEHO PROFITU z ind {def_profit} {to_return=} MINIMUM:{profit_min_ind_price_value} {smer} {data['close']}")
#fallback na profit_min_ind_price_value
to_return = profit_min_ind_price_value
state.ilog(lvl=1,e=f"PROFIT z indikatoru {def_profit} {to_return=}")
return to_return
##based on tick a direction, returns normalized prfoit price (LONG = avgp(nebo currprice)+norm.tick, SHORT=avgp(or currprice)-norm.tick)
def get_normalized_profitprice_from_tick(state, data, tick, direction: TradeDirection):
normalized_tick = normalize_tick(state, data, float(tick))
base_price = state.avgp if state.avgp != 0 else data["close"]
returned_price = price2dec(float(base_price)+normalized_tick,3) if direction == TradeDirection.LONG else price2dec(float(base_price)-normalized_tick,3)
state.ilog(lvl=0,e=f"NORMALIZED TICK {tick=} {normalized_tick=} NORM.PRICE {returned_price}")
return returned_price
def get_max_profit_price(state, data, direction: TradeDirection):
if direction == TradeDirection.LONG:
smer = "long"

4
v2realbot/strategyblocks/indicators/RSI.py
View File

@ -4,6 +4,7 @@ from v2realbot.indicators.indicators import ema, natr, roc
from v2realbot.indicators.oscillators import rsi
from traceback import format_exc
from v2realbot.strategyblocks.indicators.helpers import get_source_series, value_or_indicator
import numpy as np
#RSI INDICATOR
# type = RSI, source = [close, vwap, hlcc4], rsi_length = [14], MA_length = int (optional), on_confirmed_only = [true, false]
@ -39,7 +40,8 @@ def populate_dynamic_RSI_indicator(data, state: StrategyState, name):
rsi_length = delka
rsi_res = rsi(source, rsi_length)
rsi_value = round(rsi_res[-1],4)
val = rsi_res[-1] if np.isfinite(rsi_res[-1]) else 0
rsi_value = round(val,4)
state.indicators[name][-1]=rsi_value
state.ilog(lvl=0,e=f"IND {name} RSI {rsi_value}")

22
v2realbot/strategyblocks/indicators/atr.py
View File

@ -3,7 +3,8 @@ 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
from v2realbot.strategyblocks.indicators.helpers import get_source_series
from collections import defaultdict
#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):
@ -16,17 +17,28 @@ def populate_dynamic_atr_indicator(data, state: StrategyState, name):
#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))
# priceline with high/low/close (bars/daily bars)
#TODO tady jsem skoncil - dodelat
source = safe_get(options, "source", "bars")
source_dict = eval(f"state.{source}")
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:]
delka_close = len(source_dict["close"])
if atr_length > delka_close:
atr_length = delka_close
source_high = source_dict["high"][-atr_length:]
source_low = source_dict["low"][-atr_length:]
source_close = source_dict["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=}")
state.ilog(lvl=0,e=f"IND {name} on {source} 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:

10
v2realbot/strategyblocks/indicators/custom/basestats.py
View File

@ -111,6 +111,16 @@ def basestats(state, params, name):
#vracime most dominant
val = float(np.max(dominant_frequencies))
return 0, val
elif func == "histogram":
#takes only first N - items
dt = np.array(source_array)
#creates 4 buckets
bins = 4
mean_of_4th_bin = np.mean(dt[np.where(np.histogram(dt, bins)[1][3] <= dt)[0]])
if not np.isfinite(mean_of_4th_bin):
mean_of_4th_bin = 0
return 0, float(mean_of_4th_bin)
elif func == "maxima":
if len(source_array) < 3:

35
v2realbot/strategyblocks/indicators/custom/expression.py
View File

@ -3,6 +3,7 @@ import numpy as np
from rich import print as printanyway
from traceback import format_exc
import v2realbot.utils.utils as utls
from copy import deepcopy
# 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
@ -18,19 +19,43 @@ def expression(state: StrategyState, params, name):
funcName = "expression"
#indicator name
operation = utls.safe_get(params, "expression", None)
if operation is None :
return -2, "required param missing"
#list of indicators that should be converted beforehands
convertToNumpy = utls.safe_get(params, "convertToNumpy", [])
state.ilog(lvl=0,e=f"BEFORE {name}:{funcName} {operation=}", **params)
#nyni vytvarime kazdou iteraci nove numpy pole
#pro optimalizaci by slo pouzit array.array ktery umi
#sdilet s numpy pamet a nevytvari se pak kopie pole
#nevyhoda: neumi comprehensions a dalsi
#viz https://chat.openai.com/c/03bb0c1d-450e-4f0e-8036-d338692c1082
#opt by chatGPT
temp_ind_mapping = {k: np.array(v) if k in convertToNumpy else v for k, v in state.ind_mapping.items()}
# temp_ind_mapping = {}
# if len(convertToNumpy) > 0:
# #mozna msgpack ext in
# temp_ind_mapping = deepcopy(state.ind_mapping)
# for key in convertToNumpy:
# try:
# temp_ind_mapping[key] = np.array(state.ind_mapping[key])
# print(f"numpyed {key}")
# except Exception:
# pass
# if len(temp_ind_mapping) == 0:
# temp_ind_mapping = state.ind_mapping
#pro zacatek eval
val = eval(operation, {'state': state, 'np': np, 'utls': utls}, state.ind_mapping)
val = eval(operation, {'state': state, 'np': np, 'utls': utls}, temp_ind_mapping)
#printanyway(val)
if not np.isfinite(val):
val = 0
val = 0 if not np.isfinite(val) else val
#val = ne.evaluate(operation, state.ind_mapping)
state.ilog(lvl=1,e=f"IND {name}:{funcName} {operation=} res:{val}", **params)