53 lines
3.0 KiB
Python
53 lines
3.0 KiB
Python
"""Build and validate the complete setup schedule before changing the application."""
|
|
from datetime import datetime, timedelta
|
|
|
|
from tests.config_reader import parse_product_codes
|
|
|
|
|
|
def build_plan(config, flow="configured", product_codes=None, now=None):
|
|
now = now or datetime.now()
|
|
setup = config.get("setup", {})
|
|
session = flow in ("session", "both") or (
|
|
flow == "configured" and setup.get("run_session", True)
|
|
)
|
|
token = flow in ("token", "both") or (
|
|
flow == "configured" and setup.get("run_token", True)
|
|
)
|
|
result = {"session": None, "tokens": []}
|
|
if session:
|
|
settings = setup.get("session", {})
|
|
start = now + timedelta(minutes=settings.get("start_offset_minutes", 7))
|
|
end = datetime.combine(now.date(), datetime.strptime(settings.get("end_time", "20:30"), "%H:%M").time())
|
|
if end <= start:
|
|
raise ValueError("Session end_time must be later than its calculated start time. Update config.json.")
|
|
result["session"] = (start.strftime("%d/%m/%Y %H:%M"), end.strftime("%d/%m/%Y %H:%M"))
|
|
if token:
|
|
settings = config.get("trading_token_idas")
|
|
if not settings:
|
|
raise ValueError("Configure trading_token_idas in config.json.")
|
|
codes = parse_product_codes(product_codes if product_codes is not None else settings.get("product_codes") or settings.get("product_code"))
|
|
if not codes or len(set(codes)) != len(codes):
|
|
raise ValueError("Specify nonempty, unique token product codes.")
|
|
expiry = datetime.combine(now.date(), datetime.strptime(settings.get("expiry_time", "20:30"), "%H:%M").time())
|
|
if expiry <= now:
|
|
raise ValueError("Token expiry_time is already past. Update config.json before creating tokens.")
|
|
delivery = (now + timedelta(days=settings.get("delivery_offset_days", 1))).strftime("%d/%m/%Y")
|
|
rows = []
|
|
defaults = [("ORDER_NOR", (now + timedelta(minutes=10)).strftime("%H:%M"), "18:30"),
|
|
("ORDER_EXCH", "18:30", "19:00"), ("TRADE", "19:00", "20:00")]
|
|
for name, start, end in defaults:
|
|
row = settings.get("time_sessions", {}).get(name, {})
|
|
if row.get("start_time") is not None:
|
|
start = row["start_time"]
|
|
elif row.get("start_offset_minutes") is not None:
|
|
start = (now + timedelta(minutes=row["start_offset_minutes"])).strftime("%H:%M")
|
|
end = row.get("end_time", end)
|
|
if datetime.strptime(end, "%H:%M") <= datetime.strptime(start, "%H:%M"):
|
|
raise ValueError(f"{name} end_time must be later than start_time. Update config.json.")
|
|
rows.append((name, start, end))
|
|
result["tokens"] = [dict(settings, product_code=code, delivery_date=delivery,
|
|
expiry=expiry.strftime("%d/%m/%Y %H:%M"),
|
|
session_date=now.strftime("%d/%m/%Y"), rows=rows)
|
|
for code in codes]
|
|
return result
|