114 lines
2.7 KiB
Python
114 lines
2.7 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import sqlite3
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from django.conf import settings
|
||
|
|
from django.http import HttpRequest, HttpResponse
|
||
|
|
from django.shortcuts import render
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class TrialRow:
|
||
|
|
ann_return: float | None
|
||
|
|
ann_vol: float | None
|
||
|
|
max_drawdown: float | None
|
||
|
|
sharpe: float | None
|
||
|
|
trades_per_year: float | None
|
||
|
|
ts_utc: str | None
|
||
|
|
run_id: str | None
|
||
|
|
config_path: str | None
|
||
|
|
|
||
|
|
|
||
|
|
def _db_path() -> Path:
|
||
|
|
return Path(getattr(settings, "QFR_EXPERIMENTS_DB"))
|
||
|
|
|
||
|
|
|
||
|
|
def _connect() -> sqlite3.Connection:
|
||
|
|
con = sqlite3.connect(str(_db_path()))
|
||
|
|
con.row_factory = sqlite3.Row
|
||
|
|
return con
|
||
|
|
|
||
|
|
|
||
|
|
def index(request: HttpRequest) -> HttpResponse:
|
||
|
|
return render(
|
||
|
|
request,
|
||
|
|
"lab/index.html",
|
||
|
|
{
|
||
|
|
"db_path": str(_db_path()),
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def top_trials(request: HttpRequest) -> HttpResponse:
|
||
|
|
limit = int(request.GET.get("limit", "50"))
|
||
|
|
limit = max(10, min(500, limit))
|
||
|
|
|
||
|
|
rows: list[dict[str, Any]] = []
|
||
|
|
summary: dict[str, Any] = {}
|
||
|
|
with _connect() as con:
|
||
|
|
cur = con.execute(
|
||
|
|
"""
|
||
|
|
SELECT ann_return, ann_vol, max_drawdown, sharpe, trades_per_year, ts_utc, run_id, config_path
|
||
|
|
FROM trials
|
||
|
|
WHERE ann_return IS NOT NULL
|
||
|
|
ORDER BY ann_return DESC
|
||
|
|
LIMIT ?
|
||
|
|
""",
|
||
|
|
(limit,),
|
||
|
|
)
|
||
|
|
rows = [dict(r) for r in cur.fetchall()]
|
||
|
|
|
||
|
|
cur2 = con.execute(
|
||
|
|
"""
|
||
|
|
SELECT
|
||
|
|
COUNT(*) AS n,
|
||
|
|
MAX(ann_return) AS best_ann,
|
||
|
|
MIN(max_drawdown) AS worst_dd,
|
||
|
|
MIN(trades_per_year) AS min_tpy,
|
||
|
|
MAX(trades_per_year) AS max_tpy
|
||
|
|
FROM trials
|
||
|
|
"""
|
||
|
|
)
|
||
|
|
summary = dict(cur2.fetchone())
|
||
|
|
|
||
|
|
return render(
|
||
|
|
request,
|
||
|
|
"lab/top_trials.html",
|
||
|
|
{
|
||
|
|
"rows": rows,
|
||
|
|
"summary": summary,
|
||
|
|
"limit": limit,
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def best_artifacts(request: HttpRequest) -> HttpResponse:
|
||
|
|
limit = int(request.GET.get("limit", "50"))
|
||
|
|
limit = max(10, min(500, limit))
|
||
|
|
|
||
|
|
rows: list[dict[str, Any]] = []
|
||
|
|
with _connect() as con:
|
||
|
|
cur = con.execute(
|
||
|
|
"""
|
||
|
|
SELECT id, ts_utc, config_path, out_equity, out_weights, out_trades,
|
||
|
|
ann_return, ann_vol, max_drawdown, trades_per_year, params_json
|
||
|
|
FROM best_artifacts
|
||
|
|
ORDER BY id DESC
|
||
|
|
LIMIT ?
|
||
|
|
""",
|
||
|
|
(limit,),
|
||
|
|
)
|
||
|
|
rows = [dict(r) for r in cur.fetchall()]
|
||
|
|
|
||
|
|
return render(
|
||
|
|
request,
|
||
|
|
"lab/best_artifacts.html",
|
||
|
|
{
|
||
|
|
"rows": rows,
|
||
|
|
"limit": limit,
|
||
|
|
},
|
||
|
|
)
|