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

fix targetema begining

Browse Source
This commit is contained in:
David Brazda
2024-01-03 17:08:09 +01:00
parent b21bd9487a
commit a7df38c61b
3 changed files with 96 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
v2realbot/main.py
View File

@ -824,7 +824,7 @@ def list_models():
return {"error": "Models directory does not exist."} return {"error": "Models directory does not exist."}
# List all files in the directory # List all files in the directory
model_files = os.listdir(models_directory) model_files = sorted(os.listdir(models_directory))
return {"models": model_files} return {"models": model_files}
@app.post("/model/upload-model", dependencies=[Depends(api_key_auth)]) @app.post("/model/upload-model", dependencies=[Depends(api_key_auth)])

4
v2realbot/static/main.css
View File

@ -49,8 +49,8 @@
} }
.scrollable-div { .scrollable-div {
width: 328px; width: 478px;
height: 143px; height: 260px;
overflow-y: auto; overflow-y: auto;
border: 1px solid #585858; border: 1px solid #585858;
padding: 10px; padding: 10px;

147
v2realbot/strategyblocks/indicators/custom/targetema.py
View File

@ -18,65 +18,104 @@ Where
- end is the current position -1 - end is the current position -1
""""" """""
def targetema(state, params, name): def targetema(state, params, name):
funcName = "targetema" try:
window_length_value = safe_get(params, "window_length_value", None) funcName = "targetema"
window_length_unit= safe_get(params, "window_length_unit", "position") window_length_value = safe_get(params, "window_length_value", None)
window_length_unit= safe_get(params, "window_length_unit", "position")
downtrend, notrend, uptrend = safe_get(params, "output_vals", [-1,0,1]) downtrend, notrend, uptrend = safe_get(params, "output_vals", [-1,0,1])
source = safe_get(params, "source", None) source = safe_get(params, "source", None)
source_series = get_source_series(state, source, True) source_series = get_source_series(state, source, True)
ema_slow = safe_get(params, "ema_slow", None) ema_slow = safe_get(params, "ema_slow", None)
ema_slow_series = get_source_series(state, ema_slow, True) ema_slow_series = get_source_series(state, ema_slow, True)
ema_div = safe_get(params, "ema_div", None) ema_div = safe_get(params, "ema_div", None)
ema_div_series = get_source_series(state, ema_div) ema_div_series = get_source_series(state, ema_div)
div_pos_threshold = safe_get(params, "div_pos_threshold", None) div_pos_threshold = safe_get(params, "div_pos_threshold", None)
div_neg_threshold = safe_get(params, "div_neg_threshold", None) div_neg_threshold = safe_get(params, "div_neg_threshold", None)
#mezi start a end price musi byt tento threshold #mezi start a end price musi byt tento threshold
req_min_pct_chng = float(safe_get(params, "req_min_pct_chng", 0.04)) #required PCT chng req_min_pct_chng = float(safe_get(params, "req_min_pct_chng", 0.04)) #required PCT chng
#kvalifikuji divergence s cenou jen vyssi nez posledni (pozor pri reverse trendu musime resetovat) #kvalifikuji divergence s cenou jen vyssi nez posledni (pozor pri reverse trendu musime resetovat)
if div_pos_threshold is not None and ema_div_series[-1] > div_pos_threshold and float(source_series[-1])>params.get("last_pos",0): if div_pos_threshold is not None and ema_div_series[-1] > div_pos_threshold and float(source_series[-1])>params.get("last_pos",0):
# Finding first index where vwap is smaller than ema_slow (last cross) # Finding first index where vwap is smaller than ema_slow (last cross)
idx = np.where(source_series < ema_slow_series)[0] idx = np.where(source_series < ema_slow_series)[0]
if idx.size > 0:
#if the value on the cross has min_pct from current price to qualify
qual_price = source_series[idx[-1]] * (1 + req_min_pct_chng/100)
qualified = qual_price < source_series[-1]
if qualified:
first_idx = -len(source_series) + idx[-1]
#fill target list with 1 from crossed point until last
target_list = get_source_series(state, name)
target_list[first_idx:] = [uptrend] * abs(first_idx)
params["last_pos"] = float(source_series[-1])
return 0, notrend
elif div_neg_threshold is not None and ema_div_series[-1] < div_neg_threshold and float(source_series[-1])<params.get("last_neg",99999999):
# Finding first index where vwap is smaller than ema_slow (last cross) and price at cross must respect min PCT threshold #there is no cross yet (beginning of the market)
idx = np.where(source_series > ema_slow_series)[0] #TODO jeste to zbytecne protahuje tento signal az do doby protnuti (prvni signal je delsi a obcas zasahuje do konzolidace)
if idx.size > 0: if idx.size ==0:
#porovname zda mezi aktualni cenou a cenou v crossu je dostatecna pro kvalifikaci #if price exists lower than qualifying price
qual_price = source_series[idx[-1]] * (1 - req_min_pct_chng/100) idx = np.where(source_series*(1 + req_min_pct_chng/100) < source_series[-1])[0]
qualified = qual_price>source_series[-1]
if qualified:
first_idx = -len(source_series) + idx[-1]
#fill target list with 1 from crossed point until last
target_list = get_source_series(state, name)
target_list[first_idx:] = [downtrend] * abs(first_idx)
params["last_neg"] = float(source_series[-1])
return 0, notrend
#test resetujeme nejvyssi body po uplynuti 20 pozic od trendu (tim konci ochranne okno)
# Finding the first 1 from backwards and its position
target_numpy = get_source_series(state, name, True)
position = np.where(target_numpy[::-1] == uptrend)[0][0] + 1
if position % 20 == 0:
params["last_pos"] = 0
position = np.where(target_numpy[::-1] == downtrend)[0][0] + 1
if position % 20 == 0:
params["last_neg"] = 99999999
return 0, notrend if idx.size == 0:
pass
else:
#we are qualified, oznacime
first_idx = -len(source_series) + idx[-1]
#fill target list with 1 from crossed point until last
target_list = get_source_series(state, name)
target_list[first_idx:] = [uptrend] * abs(first_idx)
params["last_pos"] = float(source_series[-1])
return 0, notrend
#idx.size > 0:
else:
#if the value on the cross has min_pct from current price to qualify
qual_price = source_series[idx[-1]] * (1 + req_min_pct_chng/100)
qualified = qual_price < source_series[-1]
if qualified:
first_idx = -len(source_series) + idx[-1]
#fill target list with 1 from crossed point until last
target_list = get_source_series(state, name)
target_list[first_idx:] = [uptrend] * abs(first_idx)
params["last_pos"] = float(source_series[-1])
return 0, notrend
elif div_neg_threshold is not None and ema_div_series[-1] < div_neg_threshold and float(source_series[-1])<params.get("last_neg",99999999):
# Finding first index where vwap is smaller than ema_slow (last cross) and price at cross must respect min PCT threshold
idx = np.where(source_series > ema_slow_series)[0]
#there is no cross yet (beginning of the market),
if idx.size ==0:
#if price exists higher than qualifying price
idx = np.where(source_series*(1 - req_min_pct_chng/100) > source_series[-1])[0]
if idx.size == 0:
pass
else:
#we are qualified, oznacime
first_idx = -len(source_series) + idx[-1]
#fill target list with 1 from crossed point until last
target_list = get_source_series(state, name)
target_list[first_idx:] = [downtrend] * abs(first_idx)
params["last_neg"] = float(source_series[-1])
return 0, notrend
#idx.size < 0:
else:
#porovname zda mezi aktualni cenou a cenou v crossu je dostatecna pro kvalifikaci
qual_price = source_series[idx[-1]] * (1 - req_min_pct_chng/100)
qualified = qual_price>source_series[-1]
if qualified:
first_idx = -len(source_series) + idx[-1]
#fill target list with 1 from crossed point until last
target_list = get_source_series(state, name)
target_list[first_idx:] = [downtrend] * abs(first_idx)
params["last_neg"] = float(source_series[-1])
return 0, notrend
#test resetujeme nejvyssi body po uplynuti 20 pozic od trendu (tim konci ochranne okno)
# Finding the first 1 from backwards and its position
target_numpy = get_source_series(state, name, True)
position = np.where(target_numpy[::-1] == uptrend)[0][0] + 1
if position % 20 == 0:
params["last_pos"] = 0
position = np.where(target_numpy[::-1] == downtrend)[0][0] + 1
if position % 20 == 0:
params["last_neg"] = 99999999
return 0, notrend
#pri chybe vracime explciitne notrend (muze mit jinou hodnotu nez 0)
except Exception as e:
state.ilog(lvl=1,e=f"IND ERROR {name} necháváme původní", message=str(e)+format_exc())
return 0, notrend
def add_pct(pct, value): def add_pct(pct, value):
""" """