feat: TradingView-style report optimization and Python 3.9 compatibility

web_strategy_studio/backend/pyproject.toml

@@ -45,6 +45,11 @@ target-version = "py310"
 select = ["E", "F", "W", "I", "UP"]
 ignore = [
     "E501",  # line-too-long — black handles this
+    # Python 3.9 compatibility: keep Optional[T], List[T], Dict[K, V]
+    # instead of T | None, list[T], dict[K, V] (3.10+ syntax)
+    "UP006",
+    "UP035",
+    "UP045",
 ]
 
 [tool.ruff.lint.isort]

web_strategy_studio/backend/studio_api/backtest_executor.py

@@ -11,7 +11,7 @@
 import tempfile
 from datetime import date, datetime
 from pathlib import Path
-from typing import Any
+from typing import Any, Dict, Optional
 
 import pandas as pd
 
@@ -32,17 +32,17 @@ def _parse_iso(d: str) -> date:
 def _estimate_trading_fraction(done_days: int, start: date, end: date) -> float:
     """Rough progress from trading-day span when bar-level hooks are unavailable."""
     # Use pandas bdate_range (Mon-Fri) as a proxy for trading days (~250/yr)
-    # instead of calendar days (~365/yr) to avoid the ~1.46× overestimate.
+    # instead of calendar days (~365/yr) to avoid the ~1.46x overestimate.
     total = max(len(pd.bdate_range(start=start, end=end)), 1)
     return min(0.95, 0.15 + 0.75 * (done_days / total))
 
 
 async def execute_backtest(
     run_id: str,
     source_code: str,
-    params: dict[str, Any],
+    params: Dict[str, Any],
     on_log: Any = None,
-) -> dict[str, Any]:
+) -> Dict[str, Any]:
     """Run isolated subprocess; stream logs; return artifact paths or error."""
     work = Path(tempfile.mkdtemp(prefix=f"eqrun_{run_id}_"))
     artifact_sub = settings.artifact_dir / "reports" / run_id
@@ -80,7 +80,7 @@ async def execute_backtest(
         }
     )
 
-    filtered_env: dict[str, str] = {}
+    filtered_env: Dict[str, str] = {}
     for k, v in os.environ.items():
         if k in _ALLOWED_ENV_KEYS or any(k.startswith(p) for p in _ALLOWED_ENV_PREFIXES):
             filtered_env[k] = v
@@ -128,7 +128,7 @@ async def pump_stream(stream: asyncio.StreamReader, name: str) -> None:
             log_lines += 1
 
             # S5: Parse structured progress lines emitted by the engine.
-            # Format: "📍 Backtest progress: N/M (pct%)" or "Backtest progress N/M"
+            # Format: "Backtest progress: N/M (pct%)" or "Backtest progress N/M"
             # The regex handles optional emoji prefix, colon, and trailing percentage.
             m = _PROGRESS_RE.search(line)
             if m:
@@ -177,7 +177,7 @@ async def progress_tick() -> None:
     t_err = asyncio.create_task(pump_stream(proc.stderr, "stderr"))  # type: ignore[arg-type]
     t_prog = asyncio.create_task(progress_tick())
 
-    timeout_payload: dict[str, Any] | None = None
+    timeout_payload: Optional[Dict[str, Any]] = None
     try:
         await asyncio.wait_for(proc.wait(), timeout=settings.run_timeout_sec)
     except asyncio.TimeoutError:
@@ -198,7 +198,7 @@ async def progress_tick() -> None:
         return timeout_payload
 
     result_path = work / "result.json"
-    payload: dict[str, Any] = {"ok": False, "error": "No result.json", "error_code": "NO_RESULT"}
+    payload: Dict[str, Any] = {"ok": False, "error": "No result.json", "error_code": "NO_RESULT"}
     if result_path.is_file():
         try:
             payload = json.loads(result_path.read_text(encoding="utf-8"))

web_strategy_studio/backend/studio_api/completion_service.py

@@ -5,6 +5,7 @@
 import json
 from functools import lru_cache
 from pathlib import Path
+from typing import Any, Dict, List
 
 
 @lru_cache
@@ -13,14 +14,14 @@ def _symbols_path() -> Path:
 
 
 @lru_cache
-def _load_symbols() -> list[dict]:
+def _load_symbols() -> List[Dict[str, Any]]:
     p = _symbols_path()
     if not p.is_file():
         return []
     return json.loads(p.read_text(encoding="utf-8"))
 
 
-def suggest(source: str, cursor_line: int, cursor_col: int) -> list[dict]:
+def suggest(source: str, cursor_line: int, cursor_col: int) -> List[Dict[str, Any]]:
     lines = source.splitlines()
     if cursor_line < 1 or cursor_line > len(lines):
         line = ""

web_strategy_studio/backend/studio_api/config.py

@@ -1,4 +1,7 @@
+from __future__ import annotations
+
 from pathlib import Path
+from typing import List, Optional
 
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
@@ -16,7 +19,7 @@ class Settings(BaseSettings):
     )
 
     database_url: str = "sqlite+aiosqlite:///./studio.sqlite3"
-    redis_url: str | None = None  # reserved for future queue split
+    redis_url: Optional[str] = None  # reserved for future queue split
     # S11: Always resolve artifact_dir to absolute path so subprocess CWD
     # (a temp directory) doesn't break file lookups in backtest_executor.
     artifact_dir: Path = _default_repo_root() / "artifacts"
@@ -29,7 +32,7 @@ class Settings(BaseSettings):
     api_port: int = 8080
     # S1: CORS — restrict to localhost by default; override via env for staging/production.
     # Do NOT use ["*"] together with allow_credentials=True (browser spec disallows it).
-    cors_allowed_origins: list[str] = [
+    cors_allowed_origins: List[str] = [
         "http://localhost:5173",
         "http://127.0.0.1:5173",
         "http://localhost:8080",

web_strategy_studio/backend/studio_api/format_service.py

@@ -6,11 +6,12 @@
 import sys
 import tempfile
 from pathlib import Path
+from typing import Any, Dict, Optional
 
 
-def format_python(source: str, timeout: float = 30.0) -> dict:
+def format_python(source: str, timeout: float = 30.0) -> Dict[str, Any]:
     proc = None
-    tmp_path: str | None = None
+    tmp_path: Optional[str] = None
     try:
         with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False, encoding="utf-8") as f:
             f.write(source)

web_strategy_studio/backend/studio_api/isolated_runner.py

@@ -12,6 +12,7 @@
 import sys
 import traceback
 from pathlib import Path
+from typing import Optional
 
 
 def main() -> int:
@@ -125,10 +126,10 @@ def _write_result(
     work: Path,
     *,
     ok: bool,
-    html: str | None = None,
-    report_json: str | None = None,
-    error: str | None = None,
-    error_code: str | None = None,
+    html: Optional[str] = None,
+    report_json: Optional[str] = None,
+    error: Optional[str] = None,
+    error_code: Optional[str] = None,
 ) -> None:
     # Local import: avoid any accidental shadowing of the stdlib `json` module.
     import json as json_stdlib

web_strategy_studio/backend/studio_api/lint_service.py

@@ -8,15 +8,16 @@
 import sys
 import tempfile
 from pathlib import Path
+from typing import Any, Dict, List
 
 from studio_api.security_scanner import SecurityScanner, require_initialize_function
 
 PROFILE_FAST = "fast"
 PROFILE_STRICT = "strict"
 
 
-def _syntax_errors(source: str) -> list[dict]:
-    out: list[dict] = []
+def _syntax_errors(source: str) -> List[Dict[str, Any]]:
+    out: List[Dict[str, Any]] = []
     try:
         compile(source, "<strategy>", "exec", ast.PyCF_ONLY_AST)
     except SyntaxError as e:
@@ -31,7 +32,7 @@ def _syntax_errors(source: str) -> list[dict]:
     return out
 
 
-def _ruff_issues(source: str, timeout: float = 15.0) -> list[dict]:
+def _ruff_issues(source: str, timeout: float = 15.0) -> List[Dict[str, Any]]:
     with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False, encoding="utf-8") as f:
         f.write(source)
         tmp = f.name
@@ -68,15 +69,15 @@ def _ruff_issues(source: str, timeout: float = 15.0) -> list[dict]:
     return issues
 
 
-def lint_source(source: str, profile: str = PROFILE_FAST) -> dict:
+def lint_source(source: str, profile: str = PROFILE_FAST) -> Dict[str, Any]:
     syntax_errors = _syntax_errors(source)
     scanner = SecurityScanner()
     sec = scanner.scan(source)
     sec.extend(require_initialize_function(source))
 
     security_notes = [{"code": n.code, "line": n.line, "message": n.message} for n in sec]
 
-    lint_issues: list[dict] = []
+    lint_issues: List[Dict[str, Any]] = []
     if not syntax_errors:
         lint_issues = _ruff_issues(source)
 

web_strategy_studio/backend/studio_api/models.py

@@ -1,4 +1,5 @@
 from datetime import datetime, timezone
+from typing import Dict, List, Optional
 
 from sqlalchemy import JSON, DateTime, Float, ForeignKey, Integer, String, Text, func
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
@@ -16,22 +17,22 @@ class Strategy(Base):
     __tablename__ = "strategies"
 
     id: Mapped[str] = mapped_column(String(64), primary_key=True)
-    owner_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    owner_id: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
     name: Mapped[str] = mapped_column(Text)
-    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    description: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
     )
     current_version: Mapped[int] = mapped_column(Integer, default=1)
-    default_params: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+    default_params: Mapped[Optional[Dict]] = mapped_column(JSON, nullable=True)
 
-    versions: Mapped[list["StrategyVersion"]] = relationship(
+    versions: Mapped[List["StrategyVersion"]] = relationship(
         back_populates="strategy",
         cascade="all, delete-orphan",
         order_by="StrategyVersion.version",
     )
-    runs: Mapped[list["Run"]] = relationship(back_populates="strategy")
+    runs: Mapped[List["Run"]] = relationship(back_populates="strategy")
 
 
 class StrategyVersion(Base):
@@ -44,9 +45,9 @@ class StrategyVersion(Base):
     version: Mapped[int] = mapped_column(Integer)
     source_code: Mapped[str] = mapped_column(Text)
     # B4/B15: content hash for dedup; sha256 hex (64 chars) or NULL for legacy rows
-    content_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    content_hash: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
     # Named snapshot label (set by POST /snapshot)
-    label: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    label: Mapped[Optional[str]] = mapped_column(String(256), nullable=True)
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
 
     strategy: Mapped["Strategy"] = relationship(back_populates="versions")
@@ -62,14 +63,14 @@ class Run(Base):
     strategy_version: Mapped[int] = mapped_column(Integer)
     status: Mapped[str] = mapped_column(String(32), default="queued")
     progress: Mapped[float] = mapped_column(Float, default=0.0)
-    stage: Mapped[str | None] = mapped_column(String(64), nullable=True)
-    params: Mapped[dict] = mapped_column(JSON, default=dict)
-    error_code: Mapped[str | None] = mapped_column(String(64), nullable=True)
-    error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
-    html_path: Mapped[str | None] = mapped_column(Text, nullable=True)
-    json_path: Mapped[str | None] = mapped_column(Text, nullable=True)
-    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
-    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
-    worker_hostname: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    stage: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
+    params: Mapped[Dict] = mapped_column(JSON, default=dict)
+    error_code: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
+    error_message: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    html_path: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    json_path: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    started_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+    finished_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+    worker_hostname: Mapped[Optional[str]] = mapped_column(String(256), nullable=True)
 
     strategy: Mapped["Strategy"] = relationship(back_populates="runs")

web_strategy_studio/backend/studio_api/proc_registry.py

@@ -3,12 +3,9 @@
 from __future__ import annotations
 
 import asyncio
-from typing import TYPE_CHECKING
+from typing import Dict, Optional
 
-if TYPE_CHECKING:
-    pass
-
-_procs: dict[str, asyncio.subprocess.Process] = {}
+_procs: Dict[str, asyncio.subprocess.Process] = {}
 
 
 def register(run_id: str, proc: asyncio.subprocess.Process) -> None:
@@ -19,7 +16,7 @@ def unregister(run_id: str) -> None:
     _procs.pop(run_id, None)
 
 
-def get_proc(run_id: str) -> asyncio.subprocess.Process | None:
+def get_proc(run_id: str) -> Optional[asyncio.subprocess.Process]:
     """Public accessor for a live subprocess handle (B21)."""
     return _procs.get(run_id)