Add manual and offset time modes

This commit is contained in:
2026-10-13 09:45:51 +05:30
parent 771b755ca1
commit eb8a831edb
3 changed files with 252 additions and 198 deletions
+13 -1
View File
@@ -14,7 +14,9 @@
"run_session": true,
"run_token": false,
"session": {
"time_mode": "offset",
"start_offset_minutes": 5,
"start_time": "14:30",
"end_time": "20:30"
}
},
@@ -55,7 +57,9 @@
"run_token": true,
"me_instance": "52",
"session": {
"time_mode": "offset",
"start_offset_minutes": 7,
"start_time": "14:30",
"end_time": "20:30"
},
"token": {
@@ -68,7 +72,9 @@
"delivery_offset_days": 0,
"delivery_from_time": "00:00",
"delivery_to_time": "23:59",
"time_mode": "offset",
"session_start_offset_minutes": 5,
"session_start_time": "14:30",
"session_end_time": "20:00"
}
},
@@ -140,7 +146,9 @@
"delivery_offset_days": 1,
"expiry_time": "20:30",
"upc_times": {
"time_mode": "manual",
"order_normal_start_time": "13:00",
"order_normal_start_offset_minutes": 10,
"order_normal_end_time": "17:00",
"trade_session_start_time": "17:00",
"trade_session_end_time": "22:00"
@@ -171,13 +179,17 @@
"token": {
"market_code": "NATIONAL",
"product_type": "REC",
"product_codes": ["REC"],
"product_codes": [
"REC"
],
"delivery_offset_days": 1,
"expiry_time": "20:30",
"type_prefix": "REC",
"time_sessions": {
"ORDER_NOR": {
"time_mode": "offset",
"start_offset_minutes": 10,
"start_time": "14:30",
"end_time": "19:30"
},
"ORDER_EXCH": {
+10 -2
View File
@@ -13,13 +13,21 @@ from tests.locators import Locator
def get_session_window(config):
session_config = config.get("setup", {}).get("session", {})
time_mode = session_config.get("time_mode", "offset").lower()
if time_mode == "manual":
start_time = session_config.get("start_time")
if not start_time:
raise ValueError("Session time_mode 'manual' requires start_time")
elif time_mode == "offset":
start_offset_minutes = session_config.get("start_offset_minutes", 7)
start_time = (datetime.now() + timedelta(minutes=start_offset_minutes)).strftime("%H:%M")
else:
raise ValueError("Session time_mode must be 'manual' or 'offset'")
end_time = session_config.get("end_time", "20:30")
start_datetime = datetime.now() + timedelta(minutes=start_offset_minutes)
today = datetime.now().strftime("%d/%m/%Y")
return start_datetime.strftime("%d/%m/%Y %H:%M"), f"{today} {end_time}"
return f"{today} {start_time}", f"{today} {end_time}"
def create_session(driver, step, config):
+41 -7
View File
@@ -59,12 +59,16 @@ def get_delivery_date_and_expiry(token):
def get_time_session_times(token, row_text, default_start_time, default_end_time):
time_session_config = token.get("time_sessions", {}).get(row_text, {})
time_mode = time_session_config.get("time_mode")
start_time = time_session_config.get("start_time")
if start_time is None:
if time_mode == "offset":
start_offset_minutes = time_session_config.get("start_offset_minutes")
if start_offset_minutes is not None:
if start_offset_minutes is None:
raise ValueError(f"{row_text} offset mode requires start_offset_minutes")
start_time = (datetime.now() + timedelta(minutes=start_offset_minutes)).strftime("%H:%M")
elif time_mode == "manual" and not start_time:
raise ValueError(f"{row_text} manual mode requires start_time")
return (
start_time or default_start_time,
@@ -72,6 +76,29 @@ def get_time_session_times(token, row_text, default_start_time, default_end_time
)
def get_upc_order_normal_start_time(timings, now=None):
"""Resolve UPC order-normal start time from manual or offset mode."""
mode = timings.get("time_mode", "manual").lower()
if mode == "offset":
offset = timings.get("order_normal_start_offset_minutes")
if offset is None:
raise ValueError(
"UPC time_mode 'offset' requires "
"order_normal_start_offset_minutes"
)
now = datetime.now() if now is None else now
return (now + timedelta(minutes=offset)).strftime("%H:%M")
if mode != "manual":
raise ValueError("UPC time_mode must be 'manual' or 'offset'")
start_time = timings.get("order_normal_start_time")
if not start_time:
raise ValueError(
"UPC time_mode 'manual' requires order_normal_start_time"
)
return start_time
def get_product_codes(token):
product_codes = parse_product_codes(token.get("create_product_codes", token.get("product_codes")))
@@ -203,10 +230,19 @@ def open_rtm_new_token(driver, step, token):
def get_rtm_token_times(token, now=None):
now = datetime.now() if now is None else now
delivery_date = (now + timedelta(days=token.get("delivery_offset_days", 0))).strftime("%d/%m/%Y")
time_mode = token.get("time_mode", "offset").lower()
if time_mode == "manual":
session_start_time = token.get("session_start_time")
if not session_start_time:
raise ValueError("RTM time_mode 'manual' requires session_start_time")
elif time_mode == "offset":
session_start_time = (now + timedelta(minutes=token.get("session_start_offset_minutes", 5))).strftime("%H:%M")
else:
raise ValueError("RTM time_mode must be 'manual' or 'offset'")
return (
f"{delivery_date} {token.get('delivery_from_time', '00:00')}",
f"{delivery_date} {token.get('delivery_to_time', '23:59')}",
(now + timedelta(minutes=token.get("session_start_offset_minutes", 5))).strftime("%H:%M"),
session_start_time,
token.get("session_end_time", "20:00"),
)
@@ -240,7 +276,7 @@ def create_upc_token(driver, step, token):
step.click((By.XPATH, Locator.addnewtokenbtn), "click_add_new_token_upc")
timings = token.get("upc_times", {})
for name, locator, value in (
("order_normal_start", Locator.upcorderstart, timings.get("order_normal_start_time")),
("order_normal_start", Locator.upcorderstart, get_upc_order_normal_start_time(timings)),
("order_normal_end", Locator.upcorderend, timings.get("order_normal_end_time")),
("trade_start", Locator.upctradestart, timings.get("trade_session_start_time")),
("trade_end", Locator.upctradeend, timings.get("trade_session_end_time")),
@@ -303,9 +339,7 @@ def create_and_activate_token(driver, step, token):
if token.get("generation_mode") == "AnyDayUPC":
step.type((By.XPATH, Locator.tokentradedate), today, f"enter_trade_date_{product_code.lower()}")
timings = token.get("upc_times", {})
normal_start = timings.get("order_normal_start_time")
if normal_start is None and timings.get("order_normal_start_offset_minutes") is not None:
normal_start = (datetime.now() + timedelta(minutes=timings["order_normal_start_offset_minutes"])).strftime("%H:%M")
normal_start = get_upc_order_normal_start_time(timings)
values = {
Locator.upcordernormalstart: normal_start,
Locator.upcordernormalend: timings.get("order_normal_end_time"),