diff --git a/README.md b/README.md index 8e1d42c3..f3df9156 100644 --- a/README.md +++ b/README.md @@ -143,6 +143,7 @@ The following CLI configurations are available. Use the [`lean config list`](#le - [`lean project-delete`](#lean-project-delete) - [`lean report`](#lean-report) - [`lean research`](#lean-research) +- [`lean ui`](#lean-ui) - [`lean whoami`](#lean-whoami) ### `lean backtest` @@ -2293,6 +2294,25 @@ Options: _See code: [lean/commands/research.py](lean/commands/research.py)_ +### `lean ui` + +Browse projects and run commands on them in an interactive terminal UI. + +``` +Usage: lean ui [OPTIONS] + + Browse projects and run commands on them in an interactive terminal UI. + + Use the arrow keys or j/k to move, / to filter, enter to pick an action and q to quit. + +Options: + --lean-config FILE The Lean configuration file that should be used (defaults to the nearest lean.json) + --verbose Enable debug logging + --help Show this message and exit. +``` + +_See code: [lean/commands/ui.py](lean/commands/ui.py)_ + ### `lean whoami` Display who is logged in. diff --git a/lean/__main__.py b/lean/__main__.py new file mode 100644 index 00000000..b87991f0 --- /dev/null +++ b/lean/__main__.py @@ -0,0 +1,18 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Allows running the CLI with `python -m lean`, which is how `lean ui` launches commands as child processes + +from lean.main import main + +main(prog_name="lean") diff --git a/lean/click.py b/lean/click.py index 1beec014..f8f87a2d 100644 --- a/lean/click.py +++ b/lean/click.py @@ -41,7 +41,7 @@ def get_whoami_message() -> str: api_client = container.api_client cli_config_manager = container.cli_config_manager - if cli_config_manager.user_id.get_value() is not None and cli_config_manager.api_token.get_value() is not None: + if cli_config_manager.has_credentials(): try: organizations = api_client.organizations.get_all() logged_in = True diff --git a/lean/commands/__init__.py b/lean/commands/__init__.py index dfa40a41..5ce29158 100644 --- a/lean/commands/__init__.py +++ b/lean/commands/__init__.py @@ -32,6 +32,7 @@ from lean.commands.research import research from lean.commands.whoami import whoami from lean.commands.gui import gui +from lean.commands.ui import ui from lean.commands.object_store import object_store from lean.commands.private_cloud import private_cloud @@ -55,5 +56,6 @@ lean.add_command(build) lean.add_command(logs) lean.add_command(gui) +lean.add_command(ui) lean.add_command(object_store) lean.add_command(private_cloud) diff --git a/lean/commands/ui.py b/lean/commands/ui.py new file mode 100644 index 00000000..6a69702e --- /dev/null +++ b/lean/commands/ui.py @@ -0,0 +1,37 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from click import command + +from lean.click import LeanCommand +from lean.container import container + + +# requires_docker makes the CLI elevate with sudo on Linux before the UI starts, the commands it runs have no terminal +@command(cls=LeanCommand, requires_lean_config=True, requires_docker=True) +def ui() -> None: + """Browse projects and run commands on them in an interactive terminal UI. + + \b + Use the arrow keys or j/k to move, / to filter, enter to pick an action and q to quit. + """ + # Textual is imported here so it does not slow down the startup of other commands + from lean.ui.app import LeanApp + + lean_config_manager = container.lean_config_manager + try: + data_dir = lean_config_manager.get_data_directory() + except Exception: + data_dir = None + + LeanApp(lean_config_manager.get_cli_root_directory(), data_dir).run() diff --git a/lean/components/api/backtest_client.py b/lean/components/api/backtest_client.py index 5a662911..5b8ce197 100644 --- a/lean/components/api/backtest_client.py +++ b/lean/components/api/backtest_client.py @@ -11,6 +11,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +from typing import Any, Dict, List, Optional + from lean.components.api.api_client import * from lean.models.api import QCBacktest @@ -18,6 +20,10 @@ class BacktestClient: """The BacktestClient class contains methods to interact with backtests/* API endpoints.""" + # How often and how long to wait for the API to prepare a backtest's chart, orders or log + _LOADING_ATTEMPTS = 30 + _LOADING_DELAY_SECONDS = 1 + def __init__(self, api_client: 'APIClient') -> None: """Creates a new BacktestClient instance. @@ -73,3 +79,69 @@ def delete(self, project_id: int, backtest_id: str) -> None: "projectId": project_id, "backtestId": backtest_id }) + + def list(self, project_id: int) -> List[Dict[str, Any]]: + """Returns the backtests of a project without their statistics. + + :param project_id: the id of the project + :return: the backtests, each holding at least its backtestId, name and created time + """ + data = self._api.post("backtests/list", {"projectId": project_id, "includeStatistics": False}) + return data.get("backtests") or [] + + def read_chart(self, project_id: int, backtest_id: str, name: str, start: int, end: int, + count: int) -> Optional[Dict[str, Any]]: + """Returns a chart of a backtest, sampled down to the given number of points. + + :param project_id: the id of the project the backtest belongs to + :param backtest_id: the id of the backtest + :param name: the name of the chart + :param start: the start of the range to read, as a unix timestamp + :param end: the end of the range to read, as a unix timestamp + :param count: the number of points to sample the chart down to + :return: the chart, or None if it is not available yet + """ + data = self._post_when_ready("backtests/chart/read", project_id, backtest_id, + {"name": name, "count": count, "start": start, "end": end}) + return data.get("chart") or None + + def read_orders(self, project_id: int, backtest_id: str, start: int, end: int) -> Dict[str, Any]: + """Returns a page of the orders of a backtest, the API serves at most 100 orders per request. + + :param project_id: the id of the project the backtest belongs to + :param backtest_id: the id of the backtest + :param start: the index of the first order to read + :param end: the index after the last order to read + :return: the page, "orders" holding the orders and "length" the total number of orders + """ + return self._post_when_ready("backtests/orders/read", project_id, backtest_id, {"start": start, "end": end}) + + def read_log(self, project_id: int, backtest_id: str, start: int, end: int) -> Dict[str, Any]: + """Returns a page of the log of a backtest, the API serves at most 200 lines per request. + + :param project_id: the id of the project the backtest belongs to + :param backtest_id: the id of the backtest + :param start: the index of the first line to read + :param end: the index after the last line to read + :return: the page, "logs" holding the lines and "length" the total number of lines + """ + return self._post_when_ready("backtests/read/log", project_id, backtest_id, + {"start": start, "end": end, "query": " "}) + + def _post_when_ready(self, endpoint: str, project_id: int, backtest_id: str, + payload: Optional[Dict[str, Any]] = None) -> Any: + """Makes a request about a backtest, retrying while the API is still preparing the data. + + The chart, orders and log endpoints prepare their data the first time it is asked for and answer + {"status": "loading", "progress": ...} until it is ready. + """ + from time import sleep + + data = {"projectId": project_id, "backtestId": backtest_id, **(payload or {})} + for attempt in range(self._LOADING_ATTEMPTS): + response = self._api.post(endpoint, data) + if response.get("status") != "loading": + return response + if attempt < self._LOADING_ATTEMPTS - 1: + sleep(self._LOADING_DELAY_SECONDS) + raise RuntimeError(f"QuantConnect is still preparing the data of {endpoint}, try again later") diff --git a/lean/components/config/cli_config_manager.py b/lean/components/config/cli_config_manager.py index 7214b059..bbbabde2 100644 --- a/lean/components/config/cli_config_manager.py +++ b/lean/components/config/cli_config_manager.py @@ -74,6 +74,13 @@ def __init__(self, general_storage: Storage, credentials_storage: Storage) -> No self.database_update_frequency ] + def has_credentials(self) -> bool: + """Returns whether a user id and API token are stored, without checking them against the API. + + :return: True if both credentials are set, False if not + """ + return self.user_id.get_value() is not None and self.api_token.get_value() is not None + def get_option_by_key(self, key: str) -> Option: """Returns the option matching the given key. diff --git a/lean/components/docker/docker_manager.py b/lean/components/docker/docker_manager.py index 887130a7..c3bc80a0 100644 --- a/lean/components/docker/docker_manager.py +++ b/lean/components/docker/docker_manager.py @@ -103,6 +103,7 @@ def run_image(self, image: DockerImage, **kwargs) -> bool: from types import FrameType from docker.errors import APIError from docker.types import Mount + from lean.constants import IS_UI_COMMAND if not self.image_installed(image): self.pull_image(image) @@ -143,6 +144,8 @@ def run_image(self, image: DockerImage, **kwargs) -> bool: detach = kwargs.pop("detach", False) is_tty = stdout.isatty() + # Without a terminal there is nobody to interrupt twice, except `lean ui` which stops its commands that way + graceful_stop = is_tty or IS_UI_COMMAND kwargs["detach"] = True kwargs["hostname"] = kwargs["hostname"] if "hostname" in kwargs else node() @@ -224,7 +227,7 @@ def kill_container(force: bool) -> None: # Kill the container on Ctrl+C def signal_handler(sig: Signals, frame: FrameType) -> None: nonlocal force_kill_next - if not is_tty or force_kill_next or kwargs["stop_signal"] == "SIGKILL": + if not graceful_stop or force_kill_next or kwargs["stop_signal"] == "SIGKILL": force_kill_current = True else: self._logger.info("Waiting 1 minute for LEAN to exit gracefully, press Ctrl+C again to force stop") diff --git a/lean/constants.py b/lean/constants.py index cbb89ab2..97e319c1 100644 --- a/lean/constants.py +++ b/lean/constants.py @@ -122,3 +122,7 @@ # Local backtests, optimizations and live deployments write their output to a directory named after their start time OUTPUT_DIRECTORY_NAME_FORMAT = "%Y-%m-%d_%H-%M-%S" + +# Set by `lean ui` in the commands it runs, which it stops with Ctrl+Break on Windows +UI_COMMAND_ENV_VAR = "LEAN_UI_COMMAND" +IS_UI_COMMAND = environ.get(UI_COMMAND_ENV_VAR) == "1" diff --git a/lean/main.py b/lean/main.py index 061c721b..c2acd57f 100644 --- a/lean/main.py +++ b/lean/main.py @@ -88,10 +88,28 @@ def _ensure_win32_available() -> None: from lean.container import container -def main() -> None: - """This function is the entrypoint when running a Lean command in a terminal.""" +def _handle_ctrl_break_as_ctrl_c() -> None: + """On Windows `lean ui` stops a command with Ctrl+Break, which must run the command's Ctrl+C cleanup.""" + from lean.constants import IS_UI_COMMAND + if not IS_UI_COMMAND: + return + from signal import SIGINT, raise_signal, signal + try: + from signal import SIGBREAK + except ImportError: + return + signal(SIGBREAK, lambda *_: raise_signal(SIGINT)) + + +def main(prog_name: str = None) -> None: + """This function is the entrypoint when running a Lean command in a terminal. + + :param prog_name: the program name shown in help and error messages, detected from the command line if not set + """ + _handle_ctrl_break_as_ctrl_c() + try: - lean.main(standalone_mode=False) + lean.main(standalone_mode=False, prog_name=prog_name) temp_manager = container.temp_manager if temp_manager.delete_temporary_directories_when_done: diff --git a/lean/models/api.py b/lean/models/api.py index 24000f3e..96097410 100644 --- a/lean/models/api.py +++ b/lean/models/api.py @@ -185,6 +185,10 @@ class QCBacktest(WrappedBaseModel): runtimeStatistics: Optional[Dict[str, str]] = None statistics: Optional[Union[Dict[str, str], List[Any]]] = None totalPerformance: Optional[Any] = None + backtestStart: Optional[Any] = None + backtestEnd: Optional[Any] = None + charts: Optional[Union[Dict[str, Any], List[Any]]] = None + parameterSet: Optional[Any] = None def is_complete(self) -> bool: """Returns whether the backtest has completed in the cloud. diff --git a/lean/ui/__init__.py b/lean/ui/__init__.py new file mode 100644 index 00000000..9af974c0 --- /dev/null +++ b/lean/ui/__init__.py @@ -0,0 +1,12 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/lean/ui/actions.py b/lean/ui/actions.py new file mode 100644 index 00000000..3534f78a --- /dev/null +++ b/lean/ui/actions.py @@ -0,0 +1,175 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from dataclasses import dataclass +from pathlib import Path +from typing import Callable, List, Optional, Tuple + +from lean.constants import UI_COMMAND_ENV_VAR +from lean.ui.projects import UIProject +from lean.ui.results import list_local_backtests + + +@dataclass(frozen=True) +class ProjectAction: + """Something to do with a project from the UI, usually running a Lean CLI command.""" + + key: str + label: str + build_args: Optional[Callable[[UIProject], List[str]]] = None + needs_local: bool = False + needs_cloud_project: bool = False + needs_login: bool = False + refreshes_projects: bool = False + opens_results: bool = False + live: Optional[str] = None + """'local' or 'cloud' for backtests, which open their results right away instead of asking to confirm.""" + + def unavailable_reason(self, project: UIProject, logged_in: bool) -> Optional[str]: + """Returns why the action cannot run on the project, or None if it can.""" + if self.opens_results: + if list_local_backtests(project.path) or (project.is_cloud and logged_in): + return None + return "run `lean login` first" if project.is_cloud else "no backtests yet" + if self.needs_local and not project.is_local: + return "pull it first" + if self.needs_login and not logged_in: + return "run `lean login` first" + if self.needs_cloud_project and not project.is_cloud: + return "push it first" + return None + + +def _cloud_name(project: UIProject) -> str: + """Cloud commands accept the project id, which avoids local and cloud names differing.""" + return str(project.cloud_id) if project.is_cloud else project.name + + +ACTIONS = [ + ProjectAction("b", "Backtest locally", lambda p: ["backtest", p.name], needs_local=True, live="local"), + ProjectAction("c", "Backtest in the cloud", + lambda p: ["cloud", "backtest", p.name, "--push"] if p.is_local + else ["cloud", "backtest", _cloud_name(p)], + needs_login=True, refreshes_projects=True, live="cloud"), + ProjectAction("v", "View backtest results", opens_results=True), + ProjectAction("p", "Push to the cloud", lambda p: ["cloud", "push", "--project", p.name], + needs_local=True, needs_login=True, refreshes_projects=True), + ProjectAction("u", "Pull from the cloud", lambda p: ["cloud", "pull", "--project", _cloud_name(p)], + needs_cloud_project=True, needs_login=True, refreshes_projects=True), + ProjectAction("s", "Cloud live status", lambda p: ["cloud", "status", _cloud_name(p)], + needs_cloud_project=True, needs_login=True), +] + + +def parse_cloud_backtest_url(line: str) -> Optional[Tuple[int, str]]: + """Returns the project and backtest ids from the backtest url `lean cloud backtest` prints, if the line has it.""" + from re import search + match = search(r"quantconnect\.com/project/(\d+)/([0-9a-f]{32})", line) + return (int(match.group(1)), match.group(2)) if match is not None else None + + +def format_command(args: List[str]) -> str: + """Formats arguments as the `lean` command a user would type.""" + from shlex import quote + return " ".join(["lean"] + [quote(a) for a in args]) + + +class CommandProcess: + """Runs a Lean CLI command as a child process, so its console output cannot draw over the UI.""" + + def __init__(self, args: List[str], cwd: Path) -> None: + self.args = args + self._cwd = cwd + from threading import Lock + self._process = None + self._stops = 0 + self._lock = Lock() + + def run(self, on_line: Callable[[str], None]) -> int: + """Runs the command to completion, passing each line of output to on_line. Blocks the calling thread. + + :return: the exit code of the command + """ + import sys + from os import environ + from subprocess import DEVNULL, PIPE, STDOUT, Popen + + env = dict(environ, PYTHONIOENCODING="utf-8", PYTHONUNBUFFERED="1", COLUMNS="120") + env[UI_COMMAND_ENV_VAR] = "1" + creation_flags = 0 + if sys.platform == "win32": + from subprocess import CREATE_NEW_PROCESS_GROUP + creation_flags = CREATE_NEW_PROCESS_GROUP + + # In the portable (PyInstaller) build sys.executable is the lean-cli binary itself, not a Python interpreter + lean_command = [sys.executable] if getattr(sys, "frozen", False) else [sys.executable, "-m", "lean"] + + with self._lock: + self._process = Popen(lean_command + self.args, + cwd=self._cwd, + env=env, + stdin=DEVNULL, + stdout=PIPE, + stderr=STDOUT, + encoding="utf-8", + errors="replace", + creationflags=creation_flags) + if self._stops: + self._interrupt() + + for line in self._process.stdout: + on_line(line.rstrip("\r\n")) + + return self._process.wait() + + @property + def stop_requested(self) -> bool: + return self._stops > 0 + + def stop(self) -> None: + """Interrupts the command like Ctrl+C would, so it can stop its Docker containers. + + The command force stops its containers on a second interrupt, a third call kills the command itself. + Asking before the command has started interrupts it as soon as it has. + """ + with self._lock: + self._stops += 1 + if self._process is None: + return + if self._stops > 2: + if self._process.poll() is None: + self._process.kill() + else: + self._interrupt() + + def wait(self, timeout: float) -> None: + """Waits for the command to end, killing it if it takes longer than timeout seconds.""" + if self._process is None: + return + from subprocess import TimeoutExpired + try: + self._process.wait(timeout) + except TimeoutExpired: + self._process.kill() + + def _interrupt(self) -> None: + if self._process.poll() is not None: + return + + from sys import platform + if platform == "win32": + from signal import CTRL_BREAK_EVENT + self._process.send_signal(CTRL_BREAK_EVENT) + else: + from signal import SIGINT + self._process.send_signal(SIGINT) diff --git a/lean/ui/app.py b/lean/ui/app.py new file mode 100644 index 00000000..8d257caa --- /dev/null +++ b/lean/ui/app.py @@ -0,0 +1,467 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from datetime import datetime +from pathlib import Path +from typing import Deque, List, Optional + +from rich.table import Table +from rich.text import Text +from textual import on, work +from textual.app import App, ComposeResult +from textual.binding import Binding +from textual.containers import Horizontal, Vertical +from textual.screen import ModalScreen +from textual.timer import Timer +from textual.widgets import DataTable, Footer, Header, Input, OptionList, RichLog, Static +from textual.widgets.option_list import Option + +from lean.constants import OUTPUT_DIRECTORY_NAME_FORMAT +from lean.ui.actions import ACTIONS, CommandProcess, ProjectAction, format_command, parse_cloud_backtest_url +from lean.ui.backtest_screen import BacktestScreen, LiveRun +from lean.ui.projects import (UIProject, discover_local_projects, fetch_cloud_projects, filter_projects, + is_logged_in, merge_cloud_projects) +from lean.ui.results import BacktestSource, CloudBacktest, LocalBacktest, PendingCloudBacktest + +# How often the output of the running command is shown +OUTPUT_FLUSH_SECONDS = 0.1 + +# How long quitting waits for the running command to stop before killing it +QUIT_TIMEOUT_SECONDS = 10 + + +class ProjectTable(DataTable): + BINDINGS = [ + Binding("j", "cursor_down", show=False), + Binding("k", "cursor_up", show=False), + ] + + +class ActionList(OptionList): + BINDINGS = [ + Binding("j", "cursor_down", show=False), + Binding("k", "cursor_up", show=False), + ] + + +class ConfirmScreen(ModalScreen[bool]): + """Asks a yes or no question, by default whether to run the command an action is about to run.""" + + BINDINGS = [ + Binding("y,enter", "confirm", "Yes"), + Binding("n,escape", "cancel", "No"), + ] + + def __init__(self, command: str = "", question: str = "Run this command?", yes: str = "run", + no: str = "cancel") -> None: + super().__init__() + self._command = command + self._question = question + self._yes = yes + self._no = no + + def compose(self) -> ComposeResult: + message = Text.assemble((self._question, "bold"), ("\n\n", "")) + if self._command: + message.append_text(Text.assemble((self._command, "cyan"), ("\n\n", ""))) + message.append_text(Text.assemble(("y", "bold"), (f" / enter to {self._yes}, ", "dim"), + ("n", "bold"), (f" / esc to {self._no}", "dim"))) + yield Static(message, id="confirm") + + def action_confirm(self) -> None: + self.dismiss(True) + + def action_cancel(self) -> None: + self.dismiss(False) + + +class LeanApp(App): + """A keyboard-driven terminal UI for browsing projects and running Lean CLI commands on them.""" + + TITLE = "LEAN" + + CSS = """ + #sidebar { width: 2fr; min-width: 32; } + #main { width: 3fr; } + #filter { border: round $primary-darken-2; } + #filter:focus { border: round $accent; } + ProjectTable, #details, ActionList, #output { border: round $primary-darken-2; } + ProjectTable:focus, ActionList:focus, #output:focus { border: round $accent; } + ProjectTable { height: 1fr; overflow-x: hidden; } + #details { height: auto; padding: 0 1; } + ActionList { height: auto; max-height: 10; } + #output { height: 1fr; } + ConfirmScreen { align: center middle; } + #confirm { width: auto; max-width: 90%; padding: 1 2; border: round $accent; background: $surface; } + """ + + BINDINGS = [ + Binding("q", "quit", "Quit"), + Binding("slash", "focus_filter", "Filter"), + Binding("r", "refresh", "Refresh"), + Binding("x", "stop_command", "Stop command"), + Binding("escape", "back", "Back", show=False), + ] + [Binding(a.key, f"run_action('{a.key}')", a.label, show=False) for a in ACTIONS] + + def __init__(self, root: Path, data_dir: Optional[Path] = None) -> None: + super().__init__() + self._root = root + self._data_dir = data_dir + self._projects: List[UIProject] = [] + self._visible: List[UIProject] = [] + self._logged_in = False + self._process: Optional[CommandProcess] = None + self._live_screen: Optional[BacktestScreen] = None + # The command's output is read on a worker thread and shown in batches, a round trip per line stalls it + from collections import deque + self._output_lines: Deque[str] = deque() + self._output_timer: Optional[Timer] = None + + def _query(self, selector, expect_type=None): + """Queries the projects screen, which stays at the bottom of the stack while other screens are shown.""" + screen = self.screen_stack[0] + return screen.query_one(selector, expect_type) if expect_type is not None else screen.query_one(selector) + + def compose(self) -> ComposeResult: + yield Header() + with Horizontal(): + with Vertical(id="sidebar"): + yield Input(placeholder="type to filter projects", id="filter") + yield ProjectTable(cursor_type="row", zebra_stripes=True) + with Vertical(id="main"): + yield Static(id="details") + yield ActionList() + yield RichLog(id="output", wrap=True, markup=False, min_width=20) + yield Footer() + + def on_mount(self) -> None: + self.sub_title = str(self._root) + + table = self._query(ProjectTable) + table.add_columns("Where", "Lang", "Project") + table.border_title = "Projects" + table.border_subtitle = "loading projects..." + self._query("#filter", Input).border_title = "Filter (/)" + self._query("#details", Static).border_title = "Project" + self._query(ActionList).border_title = "Actions" + self._query("#output", RichLog).border_title = "Output" + + self.action_refresh() + + # Loading projects + + def action_refresh(self) -> None: + # The spinner covers the table, so a refresh keeps showing the projects it already has + if not self._projects: + self._query(ProjectTable).loading = True + self._load_projects() + + @work(thread=True, exclusive=True, group="projects") + def _load_projects(self) -> None: + skip_dirs = [self._data_dir] if self._data_dir is not None else [] + local_projects = discover_local_projects(self._root, skip_dirs) + logged_in = is_logged_in() + self.call_from_thread(self._set_projects, local_projects, logged_in, + "loading cloud projects..." if logged_in else "not logged in, showing local projects") + + if not logged_in: + return + + try: + cloud_projects = fetch_cloud_projects() + except Exception as exception: + self.call_from_thread(self._set_projects, local_projects, logged_in, "cloud projects unavailable") + self.call_from_thread(self.notify, f"Could not load cloud projects: {exception}", severity="error") + return + + projects = merge_cloud_projects(local_projects, cloud_projects) + self.call_from_thread(self._set_projects, projects, logged_in, f"{len(cloud_projects)} cloud projects") + + def _set_projects(self, projects: List[UIProject], logged_in: bool, status: str) -> None: + self._projects = projects + self._logged_in = logged_in + table = self._query(ProjectTable) + if table.loading: + # A widget cannot be focused while it is loading, so the table takes the focus once it has rows + table.loading = False + table.focus() + table.border_subtitle = status + self._apply_filter() + + # Filtering and selection + + @on(Input.Changed, "#filter") + def _filter_changed(self) -> None: + self._apply_filter() + + @on(Input.Submitted, "#filter") + def _filter_submitted(self) -> None: + self._query(ProjectTable).focus() + + def _apply_filter(self) -> None: + table = self._query(ProjectTable) + selected = self.selected_project + + self._visible = filter_projects(self._projects, self._query("#filter", Input).value) + + table.clear() + for index, project in enumerate(self._visible): + location_style = {"synced": "green", "local": "yellow", "cloud": "cyan"}[project.location] + name = Text(project.name, style="dim" if project.is_library else "") + table.add_row(Text(project.location, style=location_style), project.language, name, key=str(index)) + + table.border_title = f"Projects ({len(self._visible)}/{len(self._projects)})" + + if selected is not None: + for index, project in enumerate(self._visible): + if project.name == selected.name and project.cloud_id == selected.cloud_id: + table.move_cursor(row=index) + break + + self._show_project(self.selected_project) + + @property + def selected_project(self) -> Optional[UIProject]: + table = self._query(ProjectTable) + if not self._visible or table.cursor_row < 0 or table.cursor_row >= len(self._visible): + return None + return self._visible[table.cursor_row] + + @on(DataTable.RowHighlighted, "ProjectTable") + def _row_highlighted(self) -> None: + self._show_project(self.selected_project) + + @on(DataTable.RowSelected, "ProjectTable") + def _row_selected(self) -> None: + self._query(ActionList).focus() + + def _show_project(self, project: Optional[UIProject]) -> None: + details = self._query("#details", Static) + actions = self._query(ActionList) + actions.clear_options() + + if project is None: + query = self._query("#filter", Input).value + if self._projects and query: + message = f"No projects match '{query}'." + else: + message = "No projects found. Create one with `lean project-create`." + details.update(Text(message, style="dim")) + return + + grid = Table.grid(padding=(0, 2)) + grid.add_column(style="bold") + grid.add_column() + grid.add_row("Name", project.name) + grid.add_row("Location", project.location) + grid.add_row("Language", project.language or "unknown") + grid.add_row("Cloud id", str(project.cloud_id) if project.is_cloud else "not in the cloud") + grid.add_row("Path", str(project.path) if project.is_local else "not pulled") + if project.modified is not None: + grid.add_row("Modified", project.modified.strftime("%Y-%m-%d %H:%M")) + if project.description: + grid.add_row("Description", project.description) + details.update(grid) + + reasons = [action.unavailable_reason(project, self._logged_in) for action in ACTIONS] + for action, reason in zip(ACTIONS, reasons): + prompt = Text.assemble((f" {action.key} ", "bold reverse"), f" {action.label}") + if reason is not None: + prompt.append(f" ({reason})", style="dim") + actions.add_option(Option(prompt, id=action.key, disabled=reason is not None)) + + available = [i for i, reason in enumerate(reasons) if reason is None] + if available: + actions.highlighted = available[0] + + # Navigation + + def check_action(self, action: str, parameters: tuple) -> Optional[bool]: + # The project actions belong to the projects screen, other screens use these keys for their own bindings + if action in ("run_action", "focus_filter", "refresh") and len(self.screen_stack) > 1: + return False + return True + + def action_focus_filter(self) -> None: + self._query("#filter", Input).focus() + + def action_back(self) -> None: + filter_input = self._query("#filter", Input) + if filter_input.has_focus and filter_input.value: + filter_input.value = "" + self._query(ProjectTable).focus() + + # Running commands + + @on(OptionList.OptionSelected, "ActionList") + def _action_selected(self, event: OptionList.OptionSelected) -> None: + self.action_run_action(event.option.id) + + def action_run_action(self, key: str) -> None: + project = self.selected_project + action = next(a for a in ACTIONS if a.key == key) + if project is None: + return + + reason = action.unavailable_reason(project, self._logged_in) + if reason is not None: + self.notify(f"Cannot {action.label.lower()} '{project.name}': {reason}", severity="warning") + return + + if action.opens_results: + self.push_screen(BacktestScreen(project, self._logged_in)) + return + + if self._process is not None: + self.notify("A command is already running, press x to stop it", severity="warning") + return + + args = action.build_args(project) + + if action.live is not None: + self._start_backtest(action, project, args) + return + + def on_confirm(confirmed: Optional[bool]) -> None: + if confirmed: + self._run_command(action, args) + + self.push_screen(ConfirmScreen(format_command(args)), on_confirm) + + def _start_backtest(self, action: ProjectAction, project: UIProject, args: List[str]) -> None: + """Runs a backtest and opens its results, which fill in as the backtest runs.""" + source: BacktestSource + if action.live == "local": + # Choosing the output directory lets the results screen read it while the backtest runs + output_dir = project.path / "backtests" / datetime.now().strftime(OUTPUT_DIRECTORY_NAME_FORMAT) + args = args + ["--output", str(output_dir)] + source = LocalBacktest(output_dir) + else: + # The backtest id is known once the command has pushed and compiled the project + source = PendingCloudBacktest() + + live = LiveRun(source, lines=[f"$ {format_command(args)}"]) + self._live_screen = BacktestScreen(project, self._logged_in, live) + self.push_screen(self._live_screen) + self._run_command(action, args) + + def _run_command(self, action: ProjectAction, args: List[str]) -> None: + # Set on the UI thread so a second action cannot start before the worker does + self._process = CommandProcess(args, self._root) + self._command_started(args) + self._run_process(self._process, action.refreshes_projects) + + @work(thread=True, group="command") + def _run_process(self, process: CommandProcess, refresh_projects: bool) -> None: + try: + exit_code = process.run(self._output_lines.append) + except Exception as exception: + self._output_lines.append(f"Could not run the command: {exception}") + exit_code = -1 + + self.call_from_thread(self._command_finished, exit_code, refresh_projects) + + @property + def _live_screen_shown(self) -> Optional[BacktestScreen]: + """The results screen of the running backtest, unless it was closed.""" + return self._live_screen if self._live_screen in self.screen_stack else None + + def _command_started(self, args: List[str]) -> None: + output = self._query("#output", RichLog) + output.clear() + output.write(Text(f"$ {format_command(args)}", style="bold cyan")) + output.border_title = f"Output: {format_command(args)}" + output.border_subtitle = "running, x to stop" + self._output_lines.clear() + self._output_timer = self.set_interval(OUTPUT_FLUSH_SECONDS, self._flush_output) + + def _flush_output(self) -> None: + while self._output_lines: + self._command_output(self._output_lines.popleft()) + + def _command_output(self, line: str) -> None: + self._query("#output", RichLog).write(Text.from_ansi(line)) + + live_screen = self._live_screen_shown + if live_screen is None: + return + live_screen.add_live_line(line) + if isinstance(live_screen.live_source, PendingCloudBacktest): + ids = parse_cloud_backtest_url(line) + if ids is not None: + live_screen.set_live_source(CloudBacktest(*ids)) + + def _command_finished(self, exit_code: int, refresh_projects: bool) -> None: + self._output_timer.stop() + self._flush_output() + self._process = None + output = self._query("#output", RichLog) + output.border_subtitle = "done" if exit_code == 0 else f"failed (exit code {exit_code})" + output.write(Text(output.border_subtitle, style="green" if exit_code == 0 else "red")) + + live_screen = self._live_screen_shown + if live_screen is not None: + live_screen.add_live_line(output.border_subtitle) + live_screen.live_finished(exit_code) + self._live_screen = None + + if refresh_projects: + self.action_refresh() + else: + # A finished backtest can make the view action available + self._show_project(self.selected_project) + + def action_stop_command(self) -> None: + process = self._process + if process is None: + return + + source = self._live_screen.live_source if self._live_screen is not None else None + if not process.stop_requested and isinstance(source, CloudBacktest): + # The backtest keeps running in the cloud once the command stops, deleting it is the user's call + def on_answer(delete: Optional[bool]) -> None: + self._stop_process() + if delete: + self._delete_cloud_backtest(source) + + self.push_screen(ConfirmScreen(question="Stop watching the cloud backtest. Cancel and delete it too?", + yes="delete it", no="keep it running"), on_answer) + return + self._stop_process() + + def _stop_process(self) -> None: + if self._process is None: + return + self._process.stop() + self._query("#output", RichLog).border_subtitle = "stopping, x again to force" + + @work(thread=True, group="delete") + def _delete_cloud_backtest(self, backtest: CloudBacktest) -> None: + from lean.container import container + try: + container.api_client.backtests.delete(backtest.project_id, backtest.backtest_id) + except Exception as exception: + self.call_from_thread(self.notify, f"Could not delete the cloud backtest: {exception}", severity="error") + return + self.call_from_thread(self.notify, "Cancelled and deleted the cloud backtest") + + async def action_quit(self) -> None: + process = self._process + if process is not None: + # Exiting with the command running would hang on its thread, give it a moment to stop its containers + from asyncio import to_thread + if not process.stop_requested: + process.stop() + self._query("#output", RichLog).border_subtitle = "stopping before quitting" + await to_thread(process.wait, QUIT_TIMEOUT_SECONDS) + self.exit() diff --git a/lean/ui/backtest_screen.py b/lean/ui/backtest_screen.py new file mode 100644 index 00000000..afea0b6e --- /dev/null +++ b/lean/ui/backtest_screen.py @@ -0,0 +1,715 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import List, Optional, Sequence, Tuple + +from rich.table import Table +from rich.text import Text +from textual import on, work +from textual.app import ComposeResult +from textual.binding import Binding +from textual.containers import Vertical, VerticalScroll +from textual.events import Key +from textual.message import Message +from textual.screen import ModalScreen, Screen +from textual.timer import Timer +from textual.widget import Widget +from textual.widgets import DataTable, Footer, Header, Input, OptionList, RichLog, Static, TabbedContent, TabPane +from textual.widgets.option_list import Option +from textual.worker import get_current_worker + +from lean.ui.projects import UIProject +from lean.ui.results import (ORDERS_PAGE, BacktestResult, BacktestSource, Order, list_cloud_backtests, + list_local_backtests, sort_newest_first) +from lean.ui.ticks import format_value_ticks, time_ticks, value_ticks + +# Most equity charts are a few hundred points, this keeps rendering fast for long high resolution backtests +MAX_PLOT_POINTS = 2000 + +# Runtime statistics shown above the chart, in the order the QuantConnect UI shows them +SUMMARY_STATISTICS = ["Equity", "Net Profit", "Return", "Probabilistic Sharpe Ratio", "Unrealized", "Holdings", + "Fees", "Volume"] + +# The next page of orders loads when the cursor or the scroll position is this many rows from the end +ORDERS_LOAD_MARGIN = 20 + +# Orders can be readable a little after a cloud backtest completes, until then loading them is retried +ORDERS_RETRY_SECONDS = 10 +ORDERS_RETRIES = 6 + +# Statistics where a negative value is bad and a positive value is good +SIGNED_STATISTICS = {"Net Profit", "Return", "Unrealized", "Compounding Annual Return", "Sharpe Ratio", + "Sortino Ratio", "Alpha", "Expectancy", "Information Ratio", "Treynor Ratio"} + + +def _downsample(points: Sequence[Tuple[float, float]]) -> Sequence[Tuple[float, float]]: + if len(points) <= MAX_PLOT_POINTS: + return points + step = len(points) / MAX_PLOT_POINTS + return [points[int(i * step)] for i in range(MAX_PLOT_POINTS)] + [points[-1]] + + +def _style_statistic(name: str, value: str) -> Text: + if name in SIGNED_STATISTICS and value: + return Text(value, style="red" if value.lstrip("$").startswith("-") else "green") + return Text(value) + + +def _format_created(source: BacktestSource) -> str: + created = source.created + return created.astimezone().strftime("%Y-%m-%d %H:%M") if created is not None else "" + + +@dataclass +class LiveRun: + """A backtest the UI started, whose command output is shown as its log.""" + + source: BacktestSource + lines: List[str] = field(default_factory=list) + exit_code: Optional[int] = None + + +class EquityChart(Widget): + """The equity of a backtest over time, with dates on the x axis and dollar values on the y axis. + + When the backtest's period is known the x axis spans all of it, so a running backtest's equity grows from left + to right instead of being stretched over the full width. + """ + + DEFAULT_CSS = """ + EquityChart { width: 1fr; height: 1fr; } + """ + + def __init__(self, *, id: Optional[str] = None) -> None: + super().__init__(id=id) + self._points: Sequence[Tuple[float, float]] = [] + self._period: Optional[Tuple[float, float]] = None + self._message = "No data" + + @property + def points(self) -> Sequence[Tuple[float, float]]: + return self._points + + @property + def x_range(self) -> Optional[Tuple[float, float]]: + """The times the x axis spans, or None without points.""" + if not self._points: + return None + low, high = self._points[0][0], self._points[-1][0] + if self._period is not None: + low, high = min(low, self._period[0]), max(high, self._period[1]) + return low, high + + def set_points(self, points: Sequence[Tuple[float, float]], message: str = "No data", + period: Optional[Tuple[datetime, datetime]] = None) -> None: + """Sets the equity points, and the start and end of the backtest if known.""" + self._points = _downsample(points) + self._period = (period[0].timestamp(), period[1].timestamp()) if period is not None else None + self._message = message + self.refresh() + + def render(self) -> Text: + import plotext as plt + + figure = plt.figure + figure.clear.all() + figure.plot_size(self.size.width, self.size.height) + # The simple theme colours nothing but the data, so the widget's own colours show through + figure.theme("simple") + + if not self._points: + figure.title(self._message) + return Text.from_ansi(figure.build().string()) + + xs = [x for x, _ in self._points] + ys = [y for _, y in self._points] + equity = figure.signal(xs, ys, marker=plt.marker("braille", plt.pixel("cyan"))) + equity.lines() + figure.draw(equity) + + y_ticks, y_step = value_ticks(min(ys), max(ys), max(3, min(8, self.size.height // 2))) + figure.ruler(axis=1, side=0).ticks(y_ticks, format_value_ticks(y_ticks, y_step, "$")) + x_low, x_high = self.x_range + x_axis = figure.ruler(axis=0, side=0) + if x_high > x_low: + x_axis.lim(x_low, x_high) + x_ticks, x_labels = time_ticks(x_low, x_high, max(2, self.size.width // 16)) + x_axis.ticks(x_ticks, x_labels) + + return Text.from_ansi(figure.build().string()) + + +class OrdersTable(DataTable): + """The orders of a backtest, which asks for more orders when the cursor or scroll position nears the end.""" + + class NearEnd(Message): + pass + + def watch_scroll_y(self, old_value: float, new_value: float) -> None: + super().watch_scroll_y(old_value, new_value) + self._check_near_end() + + def watch_cursor_coordinate(self, old_coordinate, new_coordinate) -> None: + super().watch_cursor_coordinate(old_coordinate, new_coordinate) + self._check_near_end() + + @property + def near_end(self) -> bool: + """Whether the cursor or the bottom of the view is close to the last row.""" + if self.row_count == 0: + return False + # Measured in rows, as the scroll limits are not updated yet when rows have just been added + last_visible_row = int(self.scroll_y) + self.size.height + return max(last_visible_row, self.cursor_row) >= self.row_count - ORDERS_LOAD_MARGIN + + def _check_near_end(self) -> None: + if self.near_end: + self.post_message(self.NearEnd()) + + +class BacktestPicker(ModalScreen[Optional[int]]): + """Lists a project's backtests, local and cloud, to pick one to show.""" + + BINDINGS = [Binding("escape", "dismiss(None)", "Cancel")] + + DEFAULT_CSS = """ + BacktestPicker { align: center middle; } + BacktestPicker > Vertical { width: 90; max-width: 95%; height: 80%; border: round $accent; background: $surface; } + BacktestPicker OptionList { height: 1fr; border: none; } + """ + + def __init__(self, backtests: List[BacktestSource], selected: int) -> None: + super().__init__() + self._backtests = backtests + self._selected = selected + + def compose(self) -> ComposeResult: + with Vertical(): + yield Input(placeholder="type to filter backtests") + yield OptionList() + + def on_mount(self) -> None: + self.query_one(Vertical).border_title = f"Backtests ({len(self._backtests)})" + self._fill("") + self.query_one(Input).focus() + + def _fill(self, query: str) -> None: + options = self.query_one(OptionList) + options.clear_options() + words = query.casefold().split() + for index, backtest in enumerate(self._backtests): + if all(w in f"{backtest.name} {backtest.location}".casefold() for w in words): + options.add_option(Option(Text.assemble( + (f"{backtest.location:<6}", "cyan" if backtest.location == "cloud" else "yellow"), " ", + (f"{_format_created(backtest):<17}", "dim"), " ", backtest.name), id=str(index))) + ids = [options.get_option_at_index(i).id for i in range(options.option_count)] + if ids: + options.highlighted = ids.index(str(self._selected)) if str(self._selected) in ids else 0 + + @on(Input.Changed) + def _filter_changed(self, event: Input.Changed) -> None: + self._fill(event.value) + + @on(Input.Submitted) + def _filter_submitted(self) -> None: + options = self.query_one(OptionList) + if options.highlighted is not None: + self.dismiss(int(options.get_option_at_index(options.highlighted).id)) + + def on_key(self, event: Key) -> None: + # The arrow keys move through the list while typing in the filter + if event.key in ("up", "down"): + options = self.query_one(OptionList) + if event.key == "up": + options.action_cursor_up() + else: + options.action_cursor_down() + event.stop() + + @on(OptionList.OptionSelected) + def _option_selected(self, event: OptionList.OptionSelected) -> None: + self.dismiss(int(event.option.id)) + + +class BacktestScreen(Screen): + """Shows a project's backtest results like the QuantConnect.com backtest page, refreshing while they change.""" + + BINDINGS = [ + Binding("escape", "app.pop_screen", "Back"), + Binding("o", "pick_backtest", "Backtests"), + Binding("left_square_bracket", "step_backtest(-1)", "Newer"), + Binding("right_square_bracket", "step_backtest(1)", "Older"), + Binding("r", "reload", "Reload"), + Binding("1", "show_tab('statistics')", "Statistics"), + Binding("2", "show_tab('orders')", "Orders"), + Binding("3", "show_tab('logs')", "Logs"), + ] + + DEFAULT_CSS = """ + BacktestScreen #summary, BacktestScreen EquityChart, BacktestScreen TabbedContent { + border: round $primary-darken-2; + } + BacktestScreen #summary { height: auto; padding: 0 1; } + BacktestScreen EquityChart { height: 1fr; min-height: 10; } + BacktestScreen TabbedContent { height: 1fr; min-height: 10; } + BacktestScreen #statistics-scroll { height: 1fr; } + """ + + def __init__(self, project: UIProject, logged_in: bool, live: Optional[LiveRun] = None) -> None: + super().__init__() + self._project = project + self._logged_in = logged_in + self._live = live + self._backtests: List[BacktestSource] = [live.source] if live is not None else [] + self._selected = 0 + self._result: Optional[BacktestResult] = None + self._log_loaded_for: Optional[str] = None + self._poll_timer: Optional[Timer] = None + + # Orders are loaded a page at a time as they are scrolled to + self._orders_for: Optional[str] = None + self._orders_count = 0 + self._orders_total: Optional[int] = None + self._orders_loading = False + self._orders_retries = 0 + + def compose(self) -> ComposeResult: + yield Header() + with Vertical(): + yield Static(id="summary") + yield EquityChart(id="equity") + with TabbedContent(initial="statistics"): + with TabPane("Statistics", id="statistics"): + with VerticalScroll(id="statistics-scroll"): + yield Static(id="statistics-table") + with TabPane("Orders", id="orders"): + yield OrdersTable(id="orders-table", cursor_type="row", zebra_stripes=True) + with TabPane("Logs", id="logs"): + yield RichLog(id="log", markup=False, min_width=20) + yield Footer() + + def on_mount(self) -> None: + self.sub_title = self._project.name + self.query_one("#equity", EquityChart).border_title = "Strategy Equity" + self.query_one("#orders-table", DataTable).add_columns( + "ID", "Time (UTC)", "Symbol", "Type", "Side", "Quantity", "Price", "Value", "Status", "Tag") + if self._live is not None: + # Show what the command prints while the backtest starts + self.query_one(TabbedContent).active = "logs" + self._show_backtest() + self._list_backtests() + + # The list of backtests + + @property + def backtest(self) -> Optional[BacktestSource]: + return self._backtests[self._selected] if self._backtests else None + + @property + def result(self) -> Optional[BacktestResult]: + return self._result + + @work(thread=True, exclusive=True, group="list") + def _list_backtests(self) -> None: + backtests: List[BacktestSource] = list(list_local_backtests(self._project.path)) + error = None + if self._project.is_cloud and self._logged_in: + try: + backtests.extend(list_cloud_backtests(self._project.cloud_id)) + except Exception as exception: + error = exception + self.app.call_from_thread(self._set_backtests, sort_newest_first(backtests), error) + + def _set_backtests(self, backtests: List[BacktestSource], error: Optional[Exception]) -> None: + if error is not None: + self.notify(f"Could not load cloud backtests: {error}", severity="error") + + current = self.backtest + if self._live is not None: + # The backtest being run stays first, the listing may not include it yet + backtests = [self._live.source] + [b for b in backtests if b.id != self._live.source.id] + self._backtests = backtests + + if current is not None and any(b.id == current.id for b in backtests): + # Keep showing the same backtest, and the object already holding its results + self._selected = next(i for i, b in enumerate(backtests) if b.id == current.id) + self._backtests[self._selected] = current + self._update_position() + else: + self._selected = 0 + self._show_backtest() + + def action_pick_backtest(self) -> None: + if not self._backtests: + return + + def picked(index: Optional[int]) -> None: + if index is not None and index != self._selected: + self._selected = index + self._show_backtest() + + self.app.push_screen(BacktestPicker(self._backtests, self._selected), picked) + + def action_step_backtest(self, step: int) -> None: + index = self._selected + step + if 0 <= index < len(self._backtests): + self._selected = index + self._show_backtest() + + def action_show_tab(self, tab: str) -> None: + self.query_one(TabbedContent).active = tab + + def action_reload(self) -> None: + self._reset_orders() + if not self._is_live(self.backtest): + self._log_loaded_for = None + self._load() + + # The backtest the UI is running + + def set_live_source(self, source: BacktestSource) -> None: + """Replaces the backtest being run, once a new cloud backtest has an id.""" + if self._live is None: + return + current = self.backtest + old_source = self._live.source + self._live.source = source + self._backtests = [source] + [b for b in self._backtests if b is not old_source and b.id != source.id] + if current is old_source: + self._selected = 0 + self._result = None + self._log_loaded_for = source.id + if self.is_mounted: + self._update_position() + self._load() + else: + self._selected = next((i for i, b in enumerate(self._backtests) if b is current), 0) + + def add_live_line(self, line: str) -> None: + if self._live is None: + return + self._live.lines.append(line) + # The command can print before this screen is mounted, on_mount then shows what was stored + if self.is_mounted and self._is_live(self.backtest): + self.query_one("#log", RichLog).write(Text.from_ansi(line)) + + def live_finished(self, exit_code: int) -> None: + if self._live is None: + return + self._live.exit_code = exit_code + if self.is_mounted and self._is_live(self.backtest): + self._load() + + @property + def live_source(self) -> Optional[BacktestSource]: + return self._live.source if self._live is not None else None + + def _is_live(self, source: Optional[BacktestSource]) -> bool: + return self._live is not None and source is not None and source is self._live.source + + # Loading results + + def _show_backtest(self) -> None: + self._result = None + self._log_loaded_for = None + self._reset_orders() + log = self.query_one("#log", RichLog) + log.clear() + if self._is_live(self.backtest): + for line in self._live.lines: + log.write(Text.from_ansi(line)) + self._log_loaded_for = self.backtest.id + + self._update_position() + self._show_summary() + self._show_chart() + self._show_statistics() + self._load() + + def _update_position(self) -> None: + summary = self.query_one("#summary", Static) + summary.border_subtitle = f"{self._selected + 1}/{len(self._backtests)} [ ] newer/older, o all" \ + if self._backtests else "" + + def _load(self) -> None: + if self._poll_timer is not None: + self._poll_timer.stop() + self._poll_timer = None + backtest = self.backtest + if backtest is None: + self._show_summary() + return + tab = self.query_one(TabbedContent).active + self._load_result(backtest, load_log=tab == "logs" and not self._is_live(backtest)) + + @work(thread=True, exclusive=True, group="result") + def _load_result(self, backtest: BacktestSource, load_log: bool) -> None: + worker = get_current_worker() + error = None + result = log = None + try: + result = backtest.load() + if load_log: + log = backtest.load_log() + except Exception as exception: + error = exception + if not worker.is_cancelled: + self.app.call_from_thread(self._show_result, backtest, result, log, error) + + def _show_result(self, backtest: BacktestSource, result: Optional[BacktestResult], log: Optional[List[str]], + error: Optional[Exception]) -> None: + if backtest is not self.backtest: + return + if error is not None: + # Polling on would repeat the error every few seconds + self.notify(f"Could not load the backtest, press r to try again: {error}", severity="error") + return + just_finished = False + if result is not None: + just_finished = self._result is not None and not self._result.finished + self._result = result + self._apply_live_exit() + just_finished = just_finished and self._result.finished + self._show_summary() + self._show_chart() + self._show_statistics() + if log is not None: + self._show_log(log) + self._log_loaded_for = backtest.id + + if just_finished: + # Orders loaded while running may have changed status, so load them again + self._reset_orders() + self._update_orders() + + if self._should_poll(): + self._poll_timer = self.set_timer(backtest.poll_seconds, self._poll) + + def _apply_live_exit(self) -> None: + # A local run that was stopped leaves its status at Running, the end of the command ends it + if self._is_live(self.backtest) and self._live.exit_code is not None and self.backtest.location == "local": + self._result.finished = True + + def _should_poll(self) -> bool: + if self._result is not None: + return not self._result.finished + # Without results keep waiting while the backtest is being started + return self._is_live(self.backtest) and self._live.exit_code is None + + def _poll(self) -> None: + if not self._is_live(self.backtest): + self._log_loaded_for = None + self._load() + + @on(TabbedContent.TabActivated) + def _tab_activated(self, event: TabbedContent.TabActivated) -> None: + backtest = self.backtest + if backtest is None: + return + if event.pane.id == "orders": + self._update_orders() + elif event.pane.id == "logs" and self._log_loaded_for != backtest.id: + self._load() + + # Loading orders a page at a time + + def _reset_orders(self) -> None: + self._orders_for = None + self._orders_count = 0 + self._orders_total = None + self._orders_retries = 0 + self.query_one("#orders-table", OrdersTable).clear() + self._label_orders() + + def _update_orders(self) -> None: + """Loads the first page of orders when the tab is shown, and new orders while the backtest runs.""" + backtest = self.backtest + if backtest is None or self.query_one(TabbedContent).active != "orders": + return + if self._orders_for != backtest.id: + self._reset_orders() + self._orders_for = backtest.id + self._load_more_orders() + elif self._result is not None and not self._result.finished: + # Ask for any orders the running backtest has placed since the last refresh + self._load_more_orders(new_orders=True) + + @on(OrdersTable.NearEnd) + def _near_end_of_orders(self) -> None: + # Checked again, as the message may have been posted while the table was still filling + if self.query_one("#orders-table", OrdersTable).near_end: + self._load_more_orders() + + def _load_more_orders(self, new_orders: bool = False) -> None: + """Loads the next page of orders, if there are more or new_orders asks to check for new ones.""" + backtest = self.backtest + if backtest is None or self._orders_for != backtest.id or self._orders_loading: + return + if not new_orders and self._orders_total is not None and self._orders_count >= self._orders_total: + return + self._orders_loading = True + self._label_orders() + self._load_orders_page(backtest, self._orders_count) + + @work(thread=True, group="orders") + def _load_orders_page(self, backtest: BacktestSource, start: int) -> None: + try: + orders, total = backtest.load_orders(start, start + ORDERS_PAGE) + error = None + except Exception as exception: + orders, total, error = [], None, exception + self.app.call_from_thread(self._add_orders, backtest, start, orders, total, error) + + def _add_orders(self, backtest: BacktestSource, start: int, orders: List[Order], total: Optional[int], + error: Optional[Exception]) -> None: + self._orders_loading = False + self._label_orders() + if backtest is not self.backtest or self._orders_for != backtest.id or start != self._orders_count: + # The table was reset while this page loaded + if self._orders_for == getattr(self.backtest, "id", None): + self._load_more_orders() + return + if error is not None: + self.notify(f"Could not load orders: {error}", severity="error") + return + + self._show_orders(orders) + self._orders_count += len(orders) + # A running cloud backtest serves no orders yet but still reports a length, which is only trusted with orders + self._orders_total = max(total or 0, self._orders_count) if orders else self._orders_count + self._label_orders() + + if self._orders_total == 0 and self._orders_expected() and self._orders_retries < ORDERS_RETRIES: + self._orders_retries += 1 + self.set_timer(ORDERS_RETRY_SECONDS, self._retry_orders) + + def _orders_expected(self) -> bool: + """Whether the statistics count orders that the orders endpoint does not return yet.""" + result = self._result + if result is None or not result.finished: + return False + try: + return int(result.statistics.get("Total Orders", "0").replace(",", "")) > 0 + except ValueError: + return False + + def _retry_orders(self) -> None: + backtest = self.backtest + if backtest is not None and self._orders_for == backtest.id and self._orders_count == 0: + self._orders_total = None + self._load_more_orders() + + def _label_orders(self) -> None: + if self._orders_loading and self._orders_count == 0: + label = "Orders (loading...)" + elif self._orders_total is None: + label = "Orders" + elif self._orders_total == 0 and self._result is not None and not self._result.finished: + label = "Orders (none yet)" + elif self._orders_count < self._orders_total: + label = f"Orders ({self._orders_count:,} of {self._orders_total:,})" + else: + label = f"Orders ({self._orders_total:,})" + self.query_one(TabbedContent).get_tab("orders").label = label + + # Rendering + + def _status(self) -> str: + result = self._result + live = self._live if self._is_live(self.backtest) else None + if result is None: + if live is not None and live.exit_code not in (None, 0): + return f"failed (exit code {live.exit_code}), see Logs" + if live is not None and live.exit_code is None: + return "starting, see Logs" if self.backtest.location == "local" else "pushing and compiling, see Logs" + return "loading..." + status = result.status or "Running" + if not result.finished and result.progress is not None: + status = f"{status} {result.progress:.0%}" + return status + + def _show_summary(self) -> None: + summary = self.query_one("#summary", Static) + backtest = self.backtest + if backtest is None: + summary.border_title = "No backtests" + summary.update(Text("This project has no backtests yet. Press escape, then b or c to run one.", + style="dim")) + return + + name = self._result.name if self._result is not None and self._result.name else backtest.name + summary.border_title = f"{name} ({backtest.location}): {self._status()}" + + result = self._result + if result is None or not result.runtime_statistics: + summary.update(Text("Waiting for results...", style="dim")) + return + + grid = Table.grid(padding=(0, 3), expand=True) + names = [n for n in SUMMARY_STATISTICS if n in result.runtime_statistics] + for _ in names: + grid.add_column() + grid.add_row(*[Text("PSR" if n == "Probabilistic Sharpe Ratio" else n, style="dim") for n in names]) + grid.add_row(*[_style_statistic(n, result.runtime_statistics[n]) for n in names]) + if result.runtime_error: + grid.add_row(Text(result.runtime_error, style="bold red")) + summary.update(grid) + + def _show_chart(self) -> None: + result = self._result + message = "Waiting for results..." if result is None or not result.finished else "No equity data" + period = (result.start, result.end) if result is not None and result.start and result.end else None + self.query_one("#equity", EquityChart).set_points(result.equity if result is not None else [], message, + period) + + def _show_statistics(self) -> None: + table = self.query_one("#statistics-table", Static) + result = self._result + if result is None or not (result.statistics or result.parameters): + table.update(Text("Statistics are available when the backtest completes.", style="dim")) + return + + grid = Table.grid(padding=(0, 2), expand=True) + for style in ["dim", None, "dim", None]: + grid.add_column(style=style) + items = list(result.statistics.items()) + half = (len(items) + 1) // 2 + for left, right in zip(items[:half], items[half:] + [("", "")]): + grid.add_row(left[0], _style_statistic(*left), right[0], _style_statistic(*right)) + if result.parameters: + grid.add_row("", "", "", "") + grid.add_row(Text("Parameters", style="bold"), "", "", "") + for name, value in result.parameters.items(): + grid.add_row(name, str(value), "", "") + table.update(grid) + + def _show_orders(self, orders: List[Order]) -> None: + table = self.query_one("#orders-table", OrdersTable) + for order in orders: + table.add_row(str(order.id), + order.time.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") if order.time else "", + order.symbol, + order.type, + Text(order.direction, style="green" if order.direction == "Buy" else "red"), + f"{order.quantity:,.6g}", + f"{order.price:,.2f}", + f"{order.value:,.2f}", + order.status, + order.tag) + + def _show_log(self, lines: List[str]) -> None: + log = self.query_one("#log", RichLog) + log.clear() + log.write("\n".join(lines) if lines else Text("No log lines", style="dim")) diff --git a/lean/ui/projects.py b/lean/ui/projects.py new file mode 100644 index 00000000..4c3e1f0f --- /dev/null +++ b/lean/ui/projects.py @@ -0,0 +1,145 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import List, Optional + +from lean.constants import PROJECT_CONFIG_FILE_NAME +from lean.models.api import QCLanguage, QCProject + + +@dataclass +class UIProject: + """A project as shown in the UI, which may exist locally, in the cloud, or both.""" + + name: str + path: Optional[Path] = None + cloud_id: Optional[int] = None + language: str = "" + description: str = "" + modified: Optional[datetime] = None + + @property + def is_local(self) -> bool: + return self.path is not None + + @property + def is_cloud(self) -> bool: + return self.cloud_id is not None + + @property + def is_library(self) -> bool: + return self.name.split("/")[0] == "Library" + + @property + def location(self) -> str: + if self.is_local and self.is_cloud: + return "synced" + return "local" if self.is_local else "cloud" + + +def discover_local_projects(root: Path, skip_dirs: Optional[List[Path]] = None) -> List[UIProject]: + """Finds all projects under the CLI root directory. + + A project is a directory containing a project config file. Project directories are not searched further, + so backtest and live output folders inside a project are never mistaken for projects. + + :param root: the CLI root directory, the one containing the Lean config file + :param skip_dirs: directories not to search, like the data directory + :return: the local projects, sorted by name + """ + from lean.container import container + + skip = {str(p.resolve()) for p in (skip_dirs or [])} + projects = [] + + from os import walk + for current, dirs, files in walk(root): + current_path = Path(current) + if PROJECT_CONFIG_FILE_NAME in files and current_path != root: + config = container.project_config_manager.get_project_config(current_path) + # The project config names the language like the QCLanguage members, "Python" or "CSharp" + language = config.get("algorithm-language", "") + projects.append(UIProject( + name=current_path.relative_to(root).as_posix(), + path=current_path, + cloud_id=config.get("cloud-id"), + language=QCLanguage[language].value if language in QCLanguage.__members__ else "", + description=config.get("description", ""), + modified=datetime.fromtimestamp(current_path.stat().st_mtime) + )) + dirs.clear() + continue + + dirs[:] = sorted(d for d in dirs + if not d.startswith(".") and str((current_path / d).resolve()) not in skip) + + return sorted(projects, key=lambda p: p.name.casefold()) + + +def merge_cloud_projects(local_projects: List[UIProject], cloud_projects: List[QCProject]) -> List[UIProject]: + """Combines local projects with the projects in the cloud. + + Local projects are matched to cloud projects by their cloud id, cloud projects without a local copy are added. + + :param local_projects: the projects found on disk + :param cloud_projects: the projects of the working organization in the cloud + :return: all projects, sorted by name + """ + by_cloud_id = {p.cloud_id: p for p in local_projects if p.cloud_id is not None} + merged = list(local_projects) + + for cloud_project in cloud_projects: + local_project = by_cloud_id.get(cloud_project.projectId) + if local_project is not None: + local_project.description = cloud_project.description or local_project.description + local_project.modified = cloud_project.modified + continue + + merged.append(UIProject( + name=cloud_project.name, + cloud_id=cloud_project.projectId, + language=cloud_project.language.value, + description=cloud_project.description, + modified=cloud_project.modified + )) + + # Cloud projects whose id is stored locally but no longer exist are shown as local only + cloud_ids = {p.projectId for p in cloud_projects} + for project in merged: + if project.is_local and project.cloud_id not in cloud_ids: + project.cloud_id = None + + return sorted(merged, key=lambda p: p.name.casefold()) + + +def filter_projects(projects: List[UIProject], query: str) -> List[UIProject]: + """Returns the projects whose name contains every word of the query, ignoring case.""" + words = query.casefold().split() + return [p for p in projects if all(w in p.name.casefold() for w in words)] + + +def is_logged_in() -> bool: + """Returns whether credentials are stored, without contacting the API.""" + from lean.container import container + return container.cli_config_manager.has_credentials() + + +def fetch_cloud_projects() -> List[QCProject]: + """Returns the projects of the working organization in the cloud. Makes a blocking API request.""" + from lean.container import container + + organization_id = container.organization_manager.try_get_working_organization_id() + return container.api_client.projects.get_all(organization_id) diff --git a/lean/ui/results.py b/lean/ui/results.py new file mode 100644 index 00000000..ae52a4af --- /dev/null +++ b/lean/ui/results.py @@ -0,0 +1,427 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Reads LEAN backtest results from local result files and from the cloud API, which share the same JSON format +# for charts, statistics and orders. Only the Strategy Equity chart is read. + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from lean.constants import OUTPUT_DIRECTORY_NAME_FORMAT +from lean.models.api import QCBacktest + +EQUITY_CHART = "Strategy Equity" +EQUITY_SERIES = "Equity" + +# LEAN enums, see Common/Orders/OrderTypes.cs and Common/Orders/OrderEnums.cs +ORDER_TYPES = ["Market", "Limit", "Stop Market", "Stop Limit", "Market On Open", "Market On Close", + "Option Exercise", "Limit If Touched", "Combo Market", "Combo Limit", "Combo Leg Limit", + "Trailing Stop"] +ORDER_STATUSES = {0: "New", 1: "Submitted", 2: "Partially Filled", 3: "Filled", 5: "Canceled", 6: "None", + 7: "Invalid", 8: "Cancel Pending", 9: "Update Submitted"} +ORDER_DIRECTIONS = ["Buy", "Sell", "Hold"] + +# Algorithm statuses after which a backtest's results no longer change, see Common/AlgorithmStatus.cs +FINISHED_STATUSES = {"Completed", "RuntimeError", "Runtime Error", "Stopped", "Liquidated", "Deleted"} + +# Orders are loaded a page at a time, the cloud API returns at most 100 orders and 200 log lines per request +ORDERS_PAGE = 100 +CLOUD_LOG_PAGE = 200 +MAX_LOG_LINES = 2000 + +# Points to request of the equity chart from the cloud, which samples it down to this many +CLOUD_CHART_POINTS = 1000 + +# LEAN logs the date range when a backtest starts, before the result file has it +_DATA_STREAM_RANGE = r"Begin DataStream - Start: (.+?) Stop: (.+?) Time:" +_LEAN_LOG_TIME_FORMAT = "%m/%d/%Y %I:%M:%S %p" + +# Files written next to a local result file that are not the result itself +_NON_RESULT_SUFFIXES = ("-order-events.json", "-summary.json") + + +@dataclass +class Order: + id: int + time: Optional[datetime] + symbol: str + type: str + direction: str + quantity: float + price: float + value: float + status: str + tag: str + + +@dataclass +class BacktestResult: + name: str = "" + equity: List[Tuple[float, float]] = field(default_factory=list) + """(unix time, equity) points of the Strategy Equity chart, candles reduced to their closing value.""" + statistics: Dict[str, str] = field(default_factory=dict) + runtime_statistics: Dict[str, str] = field(default_factory=dict) + orders: Optional[List[Order]] = None + """The orders, or None if they are loaded separately (see BacktestSource.load_orders).""" + status: str = "" + finished: bool = False + runtime_error: str = "" + start: Optional[datetime] = None + end: Optional[datetime] = None + parameters: Dict[str, str] = field(default_factory=dict) + reported_progress: Optional[float] = None + + @property + def progress(self) -> Optional[float]: + """How far the backtest is, between 0 and 1. + + Cloud backtests report it, for local backtests it is how far the equity chart got through the date range. + """ + if self.finished: + return 1.0 + if self.reported_progress is not None: + return self.reported_progress + if not self.equity or self.start is None or self.end is None: + return None + total = (self.end - self.start).total_seconds() + done = self.equity[-1][0] - self.start.timestamp() + return min(max(done / total, 0.0), 1.0) if total > 0 else None + + +def _parse_time(value: Any) -> Optional[datetime]: + """Parses a LEAN time, either unix seconds or an ISO 8601 string, which is UTC unless it says otherwise.""" + if not value: + return None + if isinstance(value, datetime): + return value if value.tzinfo is not None else value.replace(tzinfo=timezone.utc) + if isinstance(value, (int, float)): + return datetime.fromtimestamp(value, tz=timezone.utc) + try: + time = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except ValueError: + return None + return time if time.tzinfo is not None else time.replace(tzinfo=timezone.utc) + + +def _parse_point(value: Any) -> Optional[Tuple[float, float]]: + """Parses a chart point, which is [time, value], [time, open, high, low, close] or {"x": time, "y": value}.""" + if isinstance(value, dict): + time, number = value.get("x"), value.get("y") + elif isinstance(value, list) and len(value) >= 2: + time, number = value[0], value[-1] + else: + return None + if time is None or number is None: + return None + return float(time), float(number) + + +def parse_equity(charts: Dict[str, Any]) -> List[Tuple[float, float]]: + """Returns the points of the Equity series of the Strategy Equity chart.""" + series = ((charts.get(EQUITY_CHART) or {}).get("series") or {}).get(EQUITY_SERIES) or {} + return [p for p in (_parse_point(v) for v in series.get("values") or []) if p is not None] + + +def parse_order(data: Dict[str, Any]) -> Order: + # Keys may be present with null, which must fall back like a missing key + symbol = data.get("symbol") or {} + order_type = data.get("type") or 0 + direction = data.get("direction") or 0 + return Order(id=data.get("id") or 0, + time=_parse_time(data.get("time")), + symbol=symbol.get("value", "") if isinstance(symbol, dict) else str(symbol), + type=ORDER_TYPES[order_type] if 0 <= order_type < len(ORDER_TYPES) else str(order_type), + direction=ORDER_DIRECTIONS[direction] if 0 <= direction < len(ORDER_DIRECTIONS) else str(direction), + quantity=data.get("quantity") or 0.0, + price=data.get("price") or 0.0, + value=data.get("value") or 0.0, + status=ORDER_STATUSES.get(data.get("status"), str(data.get("status"))), + tag=data.get("tag") or "") + + +def parse_orders(orders: Any) -> List[Order]: + """Parses orders, which local results store by id and the cloud API returns as a list.""" + if isinstance(orders, dict): + orders = list(orders.values()) + return sorted((parse_order(o) for o in orders or []), key=lambda o: o.id) + + +def parse_local_result(data: Dict[str, Any]) -> BacktestResult: + """Parses the result file LEAN writes while running a backtest locally.""" + state = data.get("state") or {} + configuration = data.get("algorithmConfiguration") or {} + status = state.get("Status", "") + + return BacktestResult(equity=parse_equity(data.get("charts") or {}), + statistics=data.get("statistics") or {}, + runtime_statistics=data.get("runtimeStatistics") or {}, + orders=parse_orders(data.get("orders")), + status=status or "Running", + finished=status in FINISHED_STATUSES, + runtime_error=state.get("RuntimeError", ""), + start=_parse_time(configuration.get("startDate")), + end=_parse_time(configuration.get("endDate")), + parameters=configuration.get("parameters") or {}) + + +def parse_cloud_backtest(backtest: QCBacktest, charts: Dict[str, Any]) -> BacktestResult: + """Parses a backtest from the backtests/read endpoint and its equity chart from backtests/chart/read.""" + return BacktestResult(name=backtest.name, + equity=parse_equity(charts), + statistics=backtest.statistics if isinstance(backtest.statistics, dict) else {}, + runtime_statistics=backtest.runtimeStatistics or {}, + status=backtest.status.rstrip("."), + finished=backtest.is_complete(), + runtime_error=backtest.error or "", + start=_parse_time(backtest.backtestStart), + end=_parse_time(backtest.backtestEnd), + parameters=backtest.parameterSet if isinstance(backtest.parameterSet, dict) else {}, + reported_progress=backtest.progress) + + +class BacktestSource: + """A backtest whose results can be shown, stored locally or in the cloud.""" + + location = "" + poll_seconds = 15 + + @property + def id(self) -> str: + raise NotImplementedError() + + @property + def name(self) -> str: + raise NotImplementedError() + + @property + def created(self) -> Optional[datetime]: + raise NotImplementedError() + + def load(self) -> Optional[BacktestResult]: + """Loads the summary and equity chart, returns None if there are no results yet. Blocks.""" + raise NotImplementedError() + + def load_orders(self, start: int, end: int) -> Tuple[List[Order], int]: + """Loads the orders from index start up to end, and returns them with the total number of orders.""" + raise NotImplementedError() + + def load_log(self) -> List[str]: + raise NotImplementedError() + + +class LocalBacktest(BacktestSource): + """A backtest stored in a project's backtests directory, whose result file LEAN rewrites while it runs.""" + + location = "local" + poll_seconds = 10 + + def __init__(self, path: Path) -> None: + self.path = path + self._last_result: Optional[BacktestResult] = None + + @property + def id(self) -> str: + return str(self.path) + + @property + def name(self) -> str: + return self.path.name + + @property + def created(self) -> Optional[datetime]: + try: + return datetime.strptime(self.path.name, OUTPUT_DIRECTORY_NAME_FORMAT).astimezone(timezone.utc) + except ValueError: + return None + + @property + def result_file(self) -> Optional[Path]: + """The result file, which LEAN names after the backtest id.""" + from lean.container import container + + output_id = container.output_config_manager.get_output_config(self.path).get("id") + if output_id is not None: + result_file = self.path / f"{output_id}.json" + if result_file.is_file(): + return result_file + + candidates = [f for f in self.path.glob("*.json") + if f.stem.lstrip("L-").isdigit() and not f.name.endswith(_NON_RESULT_SUFFIXES)] + return max(candidates, key=lambda f: f.stat().st_mtime) if candidates else None + + def load(self) -> Optional[BacktestResult]: + """Reads the result file, returns None if there is none yet or it is being written.""" + result_file = self.result_file + if result_file is None: + return None + from json import loads + try: + result = parse_local_result(loads(result_file.read_text(encoding="utf-8"))) + except (ValueError, OSError): + return None + result.name = self.name + if result.start is None or result.end is None: + result.start, result.end = self._date_range_from_log() + self._last_result = result + return result + + def _date_range_from_log(self) -> Tuple[Optional[datetime], Optional[datetime]]: + """Reads the backtest's date range from its log, which has it from the start of the run.""" + from re import search + + log_file = self.path / "log.txt" + if not log_file.is_file(): + return None, None + with log_file.open(encoding="utf-8", errors="replace") as file: + for line in file: + match = search(_DATA_STREAM_RANGE, line) + if match is not None: + try: + return tuple(datetime.strptime(m, _LEAN_LOG_TIME_FORMAT).replace(tzinfo=timezone.utc) + for m in match.groups()) + except ValueError: + break + return None, None + + def load_orders(self, start: int, end: int) -> Tuple[List[Order], int]: + result = self._last_result or self.load() + orders = result.orders or [] if result is not None else [] + return orders[start:end], len(orders) + + def load_log(self) -> List[str]: + log_file = self.path / "log.txt" + if not log_file.is_file(): + return [] + return log_file.read_text(encoding="utf-8", errors="replace").splitlines()[-MAX_LOG_LINES:] + + +class CloudBacktest(BacktestSource): + """A backtest in the cloud, read through the API.""" + + location = "cloud" + poll_seconds = 15 + + def __init__(self, project_id: int, backtest_id: str, name: str = "", created: Optional[datetime] = None) -> None: + self.project_id = project_id + self.backtest_id = backtest_id + self._name = name + self._created = created + + @property + def id(self) -> str: + return self.backtest_id + + @property + def name(self) -> str: + return self._name or self.backtest_id + + @property + def created(self) -> Optional[datetime]: + return self._created + + @staticmethod + def _client(): + """The BacktestClient, looked up when needed so the UI modules load without the container.""" + from lean.container import container + return container.api_client.backtests + + def load(self) -> Optional[BacktestResult]: + client = self._client() + backtest = client.get(self.project_id, self.backtest_id) + self._name = backtest.name or self._name + self._created = _parse_time(backtest.created) or self._created + + charts = {} + start, end = _parse_time(backtest.backtestStart), _parse_time(backtest.backtestEnd) + if EQUITY_CHART in (backtest.charts or {}) and start is not None and end is not None: + try: + chart = client.read_chart(self.project_id, self.backtest_id, EQUITY_CHART, + int(start.timestamp()), int(end.timestamp()), CLOUD_CHART_POINTS) + if chart: + charts[EQUITY_CHART] = chart + except Exception: + # The chart is not available until the backtest has produced some results + pass + + return parse_cloud_backtest(backtest, charts) + + def load_orders(self, start: int, end: int) -> Tuple[List[Order], int]: + page = self._client().read_orders(self.project_id, self.backtest_id, start, min(end, start + ORDERS_PAGE)) + return parse_orders(page.get("orders") or []), page.get("length", 0) + + def load_log(self) -> List[str]: + client = self._client() + first = client.read_log(self.project_id, self.backtest_id, 0, CLOUD_LOG_PAGE) + length = first.get("length", 0) + start = max(0, length - MAX_LOG_LINES) + lines: List[str] = list(first.get("logs") or []) if start == 0 else [] + position = len(lines) if start == 0 else start + while position < length: + page = client.read_log(self.project_id, self.backtest_id, position, min(position + CLOUD_LOG_PAGE, length)) + if not page.get("logs"): + break + lines.extend(page["logs"]) + position += len(page["logs"]) + return lines + + +class PendingCloudBacktest(BacktestSource): + """A cloud backtest being pushed and compiled, which has no id or results yet.""" + + location = "cloud" + + @property + def id(self) -> str: + return "pending" + + @property + def name(self) -> str: + return "New cloud backtest" + + @property + def created(self) -> Optional[datetime]: + return datetime.now(timezone.utc) + + def load(self) -> Optional[BacktestResult]: + return None + + def load_orders(self, start: int, end: int) -> Tuple[List[Order], int]: + return [], 0 + + def load_log(self) -> List[str]: + return [] + + +def list_local_backtests(project_dir: Optional[Path]) -> List[LocalBacktest]: + """Returns the backtests of a local project, newest first.""" + backtests_dir = project_dir / "backtests" if project_dir is not None else None + if backtests_dir is None or not backtests_dir.is_dir(): + return [] + return sorted((LocalBacktest(d) for d in backtests_dir.iterdir() if d.is_dir()), + key=lambda b: b.name, reverse=True) + + +def list_cloud_backtests(project_id: int) -> List[CloudBacktest]: + """Returns the backtests of a cloud project, newest first. Makes a blocking API request.""" + from lean.container import container + + backtests = [CloudBacktest(project_id, b["backtestId"], b.get("name", ""), _parse_time(b.get("created"))) + for b in container.api_client.backtests.list(project_id)] + return sorted(backtests, key=lambda b: b.created or datetime.min.replace(tzinfo=timezone.utc), reverse=True) + + +def sort_newest_first(backtests: List[BacktestSource]) -> List[BacktestSource]: + oldest = datetime.min.replace(tzinfo=timezone.utc) + return sorted(backtests, key=lambda b: b.created or oldest, reverse=True) diff --git a/lean/ui/ticks.py b/lean/ui/ticks.py new file mode 100644 index 00000000..cedc3d37 --- /dev/null +++ b/lean/ui/ticks.py @@ -0,0 +1,97 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Chooses chart axis ticks at round values and round times, the way charting libraries do + +from datetime import datetime, timezone +from math import ceil, floor, log10 +from typing import List, Tuple + +_MINUTE = 60 +_HOUR = 60 * _MINUTE +_DAY = 24 * _HOUR + +# (step, step is in months rather than seconds, label format) +_TIME_STEPS = [(s * _MINUTE, False, "%H:%M") for s in (1, 5, 15, 30)] + \ + [(s * _HOUR, False, "%m-%d %H:%M") for s in (1, 2, 4, 6, 12)] + \ + [(s * _DAY, False, "%Y-%m-%d") for s in (1, 2, 7, 14)] + \ + [(s, True, "%Y-%m") for s in (1, 2, 3, 6)] + \ + [(s * 12, True, "%Y") for s in (1, 2, 5, 10, 20, 50)] + + +def value_ticks(low: float, high: float, max_ticks: int) -> Tuple[List[float], float]: + """Returns ticks at round values (1, 2, 2.5 or 5 times a power of ten) covering the range, and their step.""" + if high <= low: + return [low], 0.0 + + # Use the smallest round step that keeps the number of ticks within the limit + magnitude = 10 ** floor(log10((high - low) / max(1, max_ticks))) + for step in (m * magnitude * scale for scale in (1, 10, 100) for m in (1, 2, 2.5, 5)): + first = ceil(low / step - 1e-9) + last = floor(high / step + 1e-9) + if 0 < last - first + 1 <= max_ticks: + return [0.0 if i == 0 else i * step for i in range(first, last + 1)], step + + # The range is narrower than a round step, so label its ends instead + return [low, high], high - low + + +def format_value_ticks(ticks: List[float], step: float, unit: str = "") -> List[str]: + """Formats ticks compactly with the precision their step needs, like $1.25M or 12.5%.""" + largest = max(abs(t) for t in ticks) + divisor, suffix = next(((d, s) for d, s in [(1e9, "B"), (1e6, "M"), (1e3, "k")] if largest >= d), (1, "")) + # The fewest decimals that show the step exactly, so 0.25 steps get 2 decimals and 5 steps get none + scaled_step = step / divisor + decimals = next((d for d in range(7) if abs(round(scaled_step, d) - scaled_step) < scaled_step * 1e-6), 2) \ + if scaled_step > 0 else 2 + + labels = [] + for tick in ticks: + text = f"{abs(tick) / divisor:,.{decimals}f}{suffix}" + sign = "-" if tick < 0 else "" + labels.append(f"{sign}${text}" if unit == "$" else f"{sign}{text}%" if unit == "%" else f"{sign}{text}") + return labels + + +def _add_months(time: datetime, months: int) -> datetime: + month_index = time.year * 12 + time.month - 1 + months + return time.replace(year=month_index // 12, month=month_index % 12 + 1) + + +def time_ticks(low: float, high: float, max_ticks: int) -> Tuple[List[float], List[str]]: + """Returns ticks at round UTC times (whole hours, days, months, years) covering the range, and their labels.""" + if high <= low: + return [low], [datetime.fromtimestamp(low, tz=timezone.utc).strftime("%Y-%m-%d")] + + for step, in_months, label_format in _TIME_STEPS: + if in_months: + start = datetime.fromtimestamp(low, tz=timezone.utc).replace(day=1, hour=0, minute=0, second=0, + microsecond=0) + start = _add_months(start, (-(start.year * 12 + start.month - 1)) % step) + if start.timestamp() < low: + start = _add_months(start, step) + times = [] + while start.timestamp() <= high and len(times) <= max_ticks: + times.append(start.timestamp()) + start = _add_months(start, step) + else: + first = ceil(low / step) * step + count = int((high - first) // step) + 1 + if count > max_ticks: + continue + times = [first + i * step for i in range(count)] + + if 0 < len(times) <= max_ticks: + return times, [datetime.fromtimestamp(t, tz=timezone.utc).strftime(label_format) for t in times] + + return [low, high], [datetime.fromtimestamp(t, tz=timezone.utc).strftime("%Y") for t in (low, high)] diff --git a/scripts/main.spec b/scripts/main.spec index 30dc230d..f8aa56df 100644 --- a/scripts/main.spec +++ b/scripts/main.spec @@ -35,9 +35,12 @@ def Entrypoint(dist, group, name, **kwargs): **kwargs ) +from PyInstaller.utils.hooks import collect_dynamic_libs + # distribution needs to be installed in the machine when trying to package. # dist, group, name -a = Entrypoint('lean', 'console_scripts', 'lean') +# plotext loads its kernel library through ctypes, which the analysis cannot see +a = Entrypoint('lean', 'console_scripts', 'lean', binaries=collect_dynamic_libs('plotext')) pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher) diff --git a/setup.py b/setup.py index 8eae6950..4d188a97 100644 --- a/setup.py +++ b/setup.py @@ -58,6 +58,8 @@ def get_stubs_version_range() -> str: "packaging", f"quantconnect-stubs{get_stubs_version_range()}", "cryptography>=41.0.4", + "textual>=1.0.0", + "plotext>=6.1.0", ] setup( diff --git a/tests/commands/test_ui.py b/tests/commands/test_ui.py new file mode 100644 index 00000000..d6ea8dd5 --- /dev/null +++ b/tests/commands/test_ui.py @@ -0,0 +1,23 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from click.testing import CliRunner + +from lean.commands import lean + + +def test_ui_requires_lean_config() -> None: + result = CliRunner().invoke(lean, ["ui"]) + + assert result.exit_code != 0 + assert "requires a Lean configuration file" in str(result.exception) diff --git a/tests/ui/__init__.py b/tests/ui/__init__.py new file mode 100644 index 00000000..9af974c0 --- /dev/null +++ b/tests/ui/__init__.py @@ -0,0 +1,12 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/tests/ui/conftest.py b/tests/ui/conftest.py new file mode 100644 index 00000000..ea3ef79b --- /dev/null +++ b/tests/ui/conftest.py @@ -0,0 +1,27 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import pytest + +from tests.conftest import initialize_container + + +@pytest.fixture(autouse=True) +def fake_filesystem(tmp_path, monkeypatch) -> None: + """Replaces the in-memory filesystem with a real temporary directory for the UI tests. + + pyfakefs switches to the real filesystem process-wide for short moments, so the threads Textual runs its + workers in sometimes read the real disk instead of the fake one. + """ + monkeypatch.chdir(tmp_path) + initialize_container() diff --git a/tests/ui/test_app.py b/tests/ui/test_app.py new file mode 100644 index 00000000..5a0b3877 --- /dev/null +++ b/tests/ui/test_app.py @@ -0,0 +1,831 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import asyncio +import json +import threading +from datetime import datetime, timezone +from pathlib import Path +from unittest import mock + +from textual.widgets import TabbedContent + +from lean.container import container +from lean.ui import app as ui_app +from lean.ui import backtest_screen +from lean.ui.actions import ACTIONS, CommandProcess, format_command, parse_cloud_backtest_url +from lean.ui.app import ConfirmScreen, LeanApp, ProjectTable +from lean.ui.backtest_screen import BacktestPicker, BacktestScreen, LiveRun +from lean.ui.projects import UIProject, discover_local_projects, filter_projects, merge_cloud_projects +from tests.test_helpers import create_api_project, create_fake_lean_cli_directory +from lean.ui.results import CloudBacktest, LocalBacktest, PendingCloudBacktest +from tests.ui.test_results import (BACKTEST_ID, create_cloud_backtest, create_cloud_orders, create_local_backtest, + create_result, mock_cloud_api) + + +def _discover() -> list: + return discover_local_projects(Path.cwd(), [Path.cwd() / "data"]) + + +def _action(key: str): + return next(a for a in ACTIONS if a.key == key) + + +def test_discover_local_projects_finds_projects_and_libraries() -> None: + create_fake_lean_cli_directory() + + names = [p.name for p in _discover()] + + assert names == ["CSharp Project", "Library/CSharp Library", "Library/Python Library", "Python Project"] + + +def test_discover_local_projects_skips_project_output_and_data_directories() -> None: + create_fake_lean_cli_directory() + for path in [Path.cwd() / "Python Project" / "backtests" / "2026-01-01_00-00-00" / "config.json", + Path.cwd() / "data" / "some-dataset" / "config.json", + Path.cwd() / ".hidden" / "config.json"]: + path.parent.mkdir(parents=True) + path.write_text("{}") + + names = [p.name for p in _discover()] + + assert names == ["CSharp Project", "Library/CSharp Library", "Library/Python Library", "Python Project"] + + +def test_discover_local_projects_reads_project_config() -> None: + create_fake_lean_cli_directory() + config_path = Path.cwd() / "Python Project" / "config.json" + config_path.write_text(json.dumps({"algorithm-language": "Python", "cloud-id": 7, "description": "Momentum"})) + + project = next(p for p in _discover() if p.name == "Python Project") + + assert project.language == "Py" + assert project.cloud_id == 7 + assert project.description == "Momentum" + assert project.path == Path.cwd() / "Python Project" + + +def test_merge_cloud_projects_matches_by_cloud_id_and_adds_cloud_only_projects() -> None: + local = [UIProject("Synced", Path("Synced"), cloud_id=1), + UIProject("Local Only", Path("Local Only")), + UIProject("Deleted In Cloud", Path("Deleted In Cloud"), cloud_id=3)] + cloud = [create_api_project(1, "Synced Cloud Name"), create_api_project(2, "Cloud Only")] + + merged = {p.name: p for p in merge_cloud_projects(local, cloud)} + + assert list(merged) == ["Cloud Only", "Deleted In Cloud", "Local Only", "Synced"] + assert merged["Synced"].location == "synced" + assert merged["Local Only"].location == "local" + assert merged["Deleted In Cloud"].location == "local" + assert merged["Cloud Only"].location == "cloud" + assert merged["Cloud Only"].cloud_id == 2 + + +def test_filter_projects_matches_every_word_ignoring_case() -> None: + projects = [UIProject("Momentum Strategy"), UIProject("Mean Reversion"), UIProject("Library/Momentum Utils")] + + assert [p.name for p in filter_projects(projects, "momentum strat")] == ["Momentum Strategy"] + assert [p.name for p in filter_projects(projects, "MOMENTUM")] == ["Momentum Strategy", "Library/Momentum Utils"] + assert len(filter_projects(projects, "")) == 3 + + +def test_actions_are_unavailable_when_project_or_login_is_missing() -> None: + local = UIProject("Local", Path("Local")) + cloud = UIProject("Cloud", cloud_id=5) + + assert _action("b").unavailable_reason(local, logged_in=False) is None + assert _action("b").unavailable_reason(cloud, logged_in=True) == "pull it first" + assert _action("c").unavailable_reason(local, logged_in=False) == "run `lean login` first" + assert _action("c").unavailable_reason(cloud, logged_in=True) is None + assert _action("p").unavailable_reason(local, logged_in=False) == "run `lean login` first" + assert _action("u").unavailable_reason(local, logged_in=True) == "push it first" + assert _action("u").unavailable_reason(cloud, logged_in=True) is None + + +def test_view_results_needs_local_backtests_or_a_cloud_project() -> None: + local = UIProject("Local", Path.cwd() / "Local") + cloud = UIProject("Cloud", cloud_id=5) + + assert _action("v").unavailable_reason(local, logged_in=True) == "no backtests yet" + assert _action("v").unavailable_reason(cloud, logged_in=False) == "run `lean login` first" + assert _action("v").unavailable_reason(cloud, logged_in=True) is None + + create_local_backtest(local.path, "2026-10-08_16-00-57", create_result()) + assert _action("v").unavailable_reason(local, logged_in=False) is None + + +def test_action_commands() -> None: + local = UIProject("My Project", Path("My Project")) + cloud = UIProject("Cloud Project", cloud_id=5) + + assert format_command(_action("b").build_args(local)) == "lean backtest 'My Project'" + assert _action("c").build_args(local) == ["cloud", "backtest", "My Project", "--push"] + assert _action("c").build_args(cloud) == ["cloud", "backtest", "5"] + assert _action("u").build_args(cloud) == ["cloud", "pull", "--project", "5"] + + +def test_parse_cloud_backtest_url_reads_the_project_and_backtest_ids() -> None: + url = f"Backtest url: https://www.quantconnect.com/project/19213997/{BACKTEST_ID}" + + assert parse_cloud_backtest_url(url) == (19213997, BACKTEST_ID) + assert parse_cloud_backtest_url("Started compiling project 'CSharp Project'") is None + + +def test_command_process_runs_the_cli_as_a_child_process() -> None: + from lean import __version__ + from lean.ui.actions import CommandProcess + + lines = [] + exit_code = CommandProcess(["--version"], Path(__file__).parent).run(lines.append) + + assert exit_code == 0 + assert lines == [f"lean {__version__}"] + + +def test_command_process_honors_a_stop_asked_before_it_started() -> None: + process = CommandProcess(["--version"], Path.cwd()) + process.stop() + + with mock.patch("subprocess.Popen") as popen: + popen.return_value.stdout = iter(["lean 0.0.0\n"]) + popen.return_value.poll.return_value = None + popen.return_value.wait.return_value = 1 + exit_code = process.run(lambda line: None) + + popen.return_value.send_signal.assert_called_once() + assert exit_code == 1 + + +def test_command_process_interrupts_twice_then_kills() -> None: + process = CommandProcess(["--version"], Path.cwd()) + + with mock.patch("subprocess.Popen") as popen: + popen.return_value.stdout = iter([]) + popen.return_value.poll.return_value = None + popen.return_value.wait.return_value = 0 + process.run(lambda line: None) + for _ in range(3): + process.stop() + + assert popen.return_value.send_signal.call_count == 2 + popen.return_value.kill.assert_called_once() + + +def test_command_process_wait_kills_a_command_that_does_not_stop() -> None: + from subprocess import TimeoutExpired + process = CommandProcess(["--version"], Path.cwd()) + + with mock.patch("subprocess.Popen") as popen: + popen.return_value.stdout = iter([]) + popen.return_value.wait.side_effect = [0, TimeoutExpired("lean", 1)] + process.run(lambda line: None) + process.wait(1) + + popen.return_value.kill.assert_called_once() + + +def _run_app(test, logged_in: bool = False) -> None: + """Runs a test coroutine against the app with the fake Lean CLI directory and no cloud projects.""" + create_fake_lean_cli_directory() + + async def run() -> None: + with mock.patch.object(ui_app, "is_logged_in", return_value=logged_in), \ + mock.patch.object(ui_app, "fetch_cloud_projects", return_value=[]): + app = LeanApp(Path.cwd(), Path.cwd() / "data") + async with app.run_test(size=(140, 40)) as pilot: + await _settle(app, pilot) + await test(app, pilot) + + asyncio.run(run()) + + +async def _settle(app: LeanApp, pilot) -> None: + for _ in range(3): + await app.workers.wait_for_complete() + await pilot.pause() + + +async def _settle_results(app: LeanApp, pilot) -> None: + """Waits for the results screen to load, while the command it runs is still going.""" + for _ in range(3): + # An empty list makes wait_for_complete wait for every worker, including the command's + result_workers = [w for w in app.workers if w.group == "result"] + if result_workers: + await app.workers.wait_for_complete(result_workers) + await pilot.pause() + + +def test_app_lists_local_projects_and_filters_them() -> None: + async def test(app: LeanApp, pilot) -> None: + table = app.query_one(ProjectTable) + assert table.row_count == 4 + + await pilot.press("slash", *"python") + assert table.row_count == 2 + assert [p.name for p in app._visible] == ["Library/Python Library", "Python Project"] + + await pilot.press("escape") + assert table.row_count == 4 + assert table.has_focus + + _run_app(test) + + +def test_app_shows_it_is_loading_until_the_projects_arrive() -> None: + create_fake_lean_cli_directory() + release = threading.Event() + + def slow_discover(root, skip_dirs): + release.wait(5) + return discover_local_projects(root, skip_dirs) + + async def run() -> None: + with mock.patch.object(ui_app, "is_logged_in", return_value=False), mock.patch.object(ui_app, "discover_local_projects", side_effect=slow_discover): + app = LeanApp(Path.cwd(), Path.cwd() / "data") + async with app.run_test(size=(140, 40)) as pilot: + await pilot.pause() + table = app.query_one(ProjectTable) + assert table.loading + assert table.border_subtitle == "loading projects..." + + release.set() + await _settle(app, pilot) + assert not table.loading + assert table.has_focus + assert table.row_count == 4 + + await pilot.press("r") + assert not table.loading + + asyncio.run(run()) + + +def test_app_moves_through_projects_with_the_keyboard() -> None: + async def test(app: LeanApp, pilot) -> None: + assert app.selected_project.name == "CSharp Project" + + await pilot.press("j", "j", "j") + assert app.selected_project.name == "Python Project" + + await pilot.press("k") + assert app.selected_project.name == "Library/Python Library" + + _run_app(test) + + +def test_app_asks_for_confirmation_before_pushing() -> None: + async def test(app: LeanApp, pilot) -> None: + with mock.patch.object(ui_app, "CommandProcess") as command_process: + await pilot.press("p") + assert isinstance(app.screen, ConfirmScreen) + assert app.screen._command == "lean cloud push --project 'CSharp Project'" + + await pilot.press("n") + assert not isinstance(app.screen, ConfirmScreen) + command_process.assert_not_called() + + _run_app(test, logged_in=True) + + +def test_app_runs_a_confirmed_command_and_shows_its_output() -> None: + async def test(app: LeanApp, pilot) -> None: + def fake_run(on_line): + on_line("Successfully pushed 'CSharp Project'") + return 0 + + with mock.patch.object(ui_app, "CommandProcess") as command_process: + command_process.return_value.run.side_effect = fake_run + await pilot.press("p", "y") + await _settle(app, pilot) + + command_process.assert_called_once_with(["cloud", "push", "--project", "CSharp Project"], Path.cwd()) + output = app.query_one("#output") + assert "Successfully pushed 'CSharp Project'" in [line.text for line in output.lines] + assert output.border_subtitle == "done" + + _run_app(test, logged_in=True) + + +def test_app_does_not_run_unavailable_actions() -> None: + async def test(app: LeanApp, pilot) -> None: + with mock.patch.object(ui_app, "CommandProcess") as command_process: + await pilot.press("p") + assert not isinstance(app.screen, ConfirmScreen) + command_process.assert_not_called() + + _run_app(test) + + +def test_quitting_waits_for_the_running_command_to_stop() -> None: + async def test(app: LeanApp, pilot) -> None: + finish = threading.Event() + + def fake_run(on_line): + finish.wait(5) + return 0 + + with mock.patch.object(ui_app, "CommandProcess") as command_process: + command_process.return_value.run.side_effect = fake_run + command_process.return_value.stop_requested = False + command_process.return_value.wait.side_effect = lambda timeout: finish.set() + await pilot.press("p", "y", "q") + await _settle(app, pilot) + + command_process.return_value.stop.assert_called_once_with() + command_process.return_value.wait.assert_called_once_with(ui_app.QUIT_TIMEOUT_SECONDS) + assert not app.is_running + + _run_app(test, logged_in=True) + + +def test_quitting_does_not_force_a_command_that_is_already_stopping() -> None: + async def test(app: LeanApp, pilot) -> None: + finish = threading.Event() + + def fake_run(on_line): + finish.wait(5) + return 0 + + with mock.patch.object(ui_app, "CommandProcess") as command_process: + command_process.return_value.run.side_effect = fake_run + command_process.return_value.stop_requested = True + command_process.return_value.wait.side_effect = lambda timeout: finish.set() + await pilot.press("p", "y", "q") + await _settle(app, pilot) + + command_process.return_value.stop.assert_not_called() + command_process.return_value.wait.assert_called_once_with(ui_app.QUIT_TIMEOUT_SECONDS) + + _run_app(test, logged_in=True) + + +def _stop_cloud_backtest(answer: str): + """Runs a cloud backtest, presses x once it has its id and answers the question, returning the API mock.""" + async def test(app: LeanApp, pilot) -> None: + mock_cloud_api(create_cloud_backtest(completed=False, progress=0.5)) + delete = container.api_client.backtests.delete = mock.Mock() + finish = threading.Event() + + def fake_run(on_line): + on_line(f"Backtest url: https://www.quantconnect.com/project/123/{BACKTEST_ID}") + finish.wait(5) + return 1 + + with mock.patch.object(ui_app, "CommandProcess") as command_process: + command_process.return_value.run.side_effect = fake_run + command_process.return_value.stop_requested = False + command_process.return_value.stop.side_effect = finish.set + await pilot.press("c") + await _settle_results(app, pilot) + await pilot.press("x") + assert isinstance(app.screen, ConfirmScreen) + await pilot.press(answer) + await _settle(app, pilot) + + command_process.return_value.stop.assert_called_once_with() + test.delete_calls = delete.call_args_list + + _run_app(test, logged_in=True) + return test.delete_calls + + +def test_stopping_a_cloud_backtest_can_delete_it() -> None: + assert _stop_cloud_backtest("y") == [mock.call(123, BACKTEST_ID)] + + +def test_stopping_a_cloud_backtest_can_keep_it_running() -> None: + assert _stop_cloud_backtest("n") == [] + + +def test_app_runs_one_command_at_a_time() -> None: + async def test(app: LeanApp, pilot) -> None: + finish = threading.Event() + + def fake_run(on_line): + finish.wait(5) + return 0 + + with mock.patch.object(ui_app, "CommandProcess") as command_process: + command_process.return_value.run.side_effect = fake_run + await pilot.press("p", "y", "b") + assert app._process is not None + finish.set() + await _settle(app, pilot) + + command_process.assert_called_once_with(["cloud", "push", "--project", "CSharp Project"], Path.cwd()) + assert app._process is None + + _run_app(test, logged_in=True) + + +def test_backtesting_locally_opens_the_results_while_the_backtest_runs() -> None: + async def test(app: LeanApp, pilot) -> None: + screen_loaded = threading.Event() + + def fake_run(on_line): + output_dir = Path(command_process.call_args.args[0][3]) + on_line("Starting LEAN") + # LEAN writes the result file after the results screen has opened without one + screen_loaded.wait(5) + create_local_backtest(output_dir.parent.parent, output_dir.name, create_result()) + return 0 + + with mock.patch.object(ui_app, "CommandProcess") as command_process: + command_process.return_value.run.side_effect = fake_run + await pilot.press("b") + assert isinstance(app.screen, BacktestScreen) + await _settle_results(app, pilot) + screen_loaded.set() + await _settle(app, pilot) + + args = command_process.call_args.args[0] + assert args[:3] == ["backtest", "CSharp Project", "--output"] + assert Path(args[3]).parent == Path.cwd() / "CSharp Project" / "backtests" + + screen = app.screen + assert screen.query_one("#summary").border_title.endswith("(local): Completed") + assert len(screen.query_one("#equity").points) == 5 + assert screen.query_one(TabbedContent).active == "logs" + assert "Starting LEAN" in [line.text for line in screen.query_one("#log").lines] + + _run_app(test) + + +def test_backtesting_in_the_cloud_switches_to_the_backtest_once_it_is_created() -> None: + async def test(app: LeanApp, pilot) -> None: + mock_cloud_api(create_cloud_backtest()) + + def fake_run(on_line): + on_line("Started compiling project 'CSharp Project'") + on_line(f"Backtest url: https://www.quantconnect.com/project/123/{BACKTEST_ID}") + return 0 + + with mock.patch.object(ui_app, "CommandProcess") as command_process: + command_process.return_value.run.side_effect = fake_run + await pilot.press("c") + assert isinstance(app.screen, BacktestScreen) + await _settle(app, pilot) + + command_process.assert_called_once_with(["cloud", "backtest", "CSharp Project", "--push"], Path.cwd()) + screen = app.screen + assert isinstance(screen.live_source, CloudBacktest) + assert (screen.live_source.project_id, screen.live_source.backtest_id) == (123, BACKTEST_ID) + assert screen.query_one("#summary").border_title == "Logical Red Monkey (cloud): Completed" + assert len(screen.query_one("#equity").points) == 5 + + _run_app(test, logged_in=True) + + +def _new_screen(*args) -> BacktestScreen: + """Builds a screen outside the app, Python 3.9 needs an event loop for the lock every widget creates.""" + async def build() -> BacktestScreen: + return BacktestScreen(*args) + return asyncio.run(build()) + + +def test_a_new_cloud_backtest_id_keeps_showing_its_live_log() -> None: + pending = PendingCloudBacktest() + screen = _new_screen(UIProject("CSharp Project"), True, LiveRun(pending, lines=["Started compiling"])) + + screen.set_live_source(CloudBacktest(123, BACKTEST_ID)) + + assert screen.backtest.id == BACKTEST_ID + assert screen._log_loaded_for == BACKTEST_ID + + +def test_a_new_cloud_backtest_id_keeps_the_backtest_the_user_moved_to() -> None: + project = UIProject("CSharp Project", path=Path.cwd() / "CSharp Project") + pending = PendingCloudBacktest() + listed = CloudBacktest(123, BACKTEST_ID) + older = LocalBacktest(Path.cwd() / "CSharp Project" / "backtests" / "2026-10-01_10-00-00") + screen = _new_screen(project, True, LiveRun(pending, lines=[])) + # The listing already holds the new backtest, so replacing the pending one shifts the older one down + screen._backtests = [pending, listed, older] + screen._selected = 2 + + screen.set_live_source(CloudBacktest(123, BACKTEST_ID)) + + assert screen.backtest is older + assert [b.id for b in screen._backtests] == [BACKTEST_ID, older.id] + + +def test_a_failed_cloud_backtest_shows_the_failure() -> None: + async def test(app: LeanApp, pilot) -> None: + def fake_run(on_line): + on_line("Build Error: main.py line 3") + return 1 + + with mock.patch.object(ui_app, "CommandProcess") as command_process: + command_process.return_value.run.side_effect = fake_run + await pilot.press("c") + await _settle(app, pilot) + + screen = app.screen + assert screen.query_one("#summary").border_title == \ + "New cloud backtest (cloud): failed (exit code 1), see Logs" + assert "Build Error: main.py line 3" in [line.text for line in screen.query_one("#log").lines] + + _run_app(test, logged_in=True) + + +def test_results_screen_keys_do_not_reach_the_projects_screen() -> None: + async def test(app: LeanApp, pilot) -> None: + project_dir = Path.cwd() / "CSharp Project" + create_local_backtest(project_dir, "2026-10-08_16-00-57", create_result()) + + await pilot.press("v") + await _settle(app, pilot) + screen = app.screen + assert isinstance(screen, BacktestScreen) + + with mock.patch.object(app, "_show_project") as show_project: + await pilot.press("2") + await _settle(app, pilot) + await pilot.press("down", "enter") + await _settle(app, pilot) + + show_project.assert_not_called() + assert app.screen is screen + + _run_app(test) + + +def test_viewing_results_shows_the_newest_backtest() -> None: + async def test(app: LeanApp, pilot) -> None: + project_dir = Path.cwd() / "CSharp Project" + create_local_backtest(project_dir, "2026-10-08_16-00-57", create_result()) + create_local_backtest(project_dir, "2026-10-09_09-30-00", create_result()) + + await pilot.press("v") + await _settle(app, pilot) + + screen = app.screen + assert isinstance(screen, BacktestScreen) + assert screen.query_one("#summary").border_title == "2026-10-09_09-30-00 (local): Completed" + + await pilot.press("2") + await _settle(app, pilot) + assert screen.query_one("#orders-table").row_count == 2 + + await pilot.press("right_square_bracket") + await _settle(app, pilot) + assert screen.query_one("#summary").border_title == "2026-10-08_16-00-57 (local): Completed" + + await pilot.press("escape") + assert not isinstance(app.screen, BacktestScreen) + + _run_app(test) + + +def test_results_list_local_and_cloud_backtests_newest_first() -> None: + async def test(app: LeanApp, pilot) -> None: + project = UIProject("Cloud Project", Path.cwd() / "Cloud Project", cloud_id=123) + create_local_backtest(project.path, "2026-05-01_12-00-00", create_result()) + mock_cloud_api(create_cloud_backtest()) + + app.push_screen(BacktestScreen(project, logged_in=True)) + await _settle(app, pilot) + + screen = app.screen + assert [(b.location, b.name) for b in screen._backtests] == \ + [("cloud", "Logical Red Monkey"), ("local", "2026-05-01_12-00-00"), ("cloud", "Older")] + assert screen.query_one("#summary").border_title == "Logical Red Monkey (cloud): Completed" + + _run_app(test, logged_in=True) + + +def test_backtest_picker_filters_and_opens_a_backtest() -> None: + async def test(app: LeanApp, pilot) -> None: + project_dir = Path.cwd() / "CSharp Project" + create_local_backtest(project_dir, "2026-10-08_16-00-57", create_result()) + create_local_backtest(project_dir, "2026-10-09_09-30-00", create_result()) + + await pilot.press("v") + await _settle(app, pilot) + await pilot.press("o") + assert isinstance(app.screen, BacktestPicker) + + await pilot.press(*"10-08", "enter") + await _settle(app, pilot) + + assert isinstance(app.screen, BacktestScreen) + assert app.screen.query_one("#summary").border_title == "2026-10-08_16-00-57 (local): Completed" + + _run_app(test) + + +def test_results_of_a_running_backtest_refresh_until_it_finishes() -> None: + async def test(app: LeanApp, pilot) -> None: + running = create_result(status="Running", equity_points=5) + equity = running["charts"]["Strategy Equity"]["series"]["Equity"] + equity["values"] = equity["values"][:3] + output_dir = create_local_backtest(Path.cwd() / "CSharp Project", "2026-10-08_16-00-57", running) + + await pilot.press("v") + await _settle(app, pilot) + screen = app.screen + assert screen.query_one("#summary").border_title.startswith("2026-10-08_16-00-57 (local): Running ") + assert screen._poll_timer is not None + + # The chart spans the whole backtest, so the partial equity only fills part of it + chart = screen.query_one("#equity") + period = (datetime(2013, 10, 7, tzinfo=timezone.utc).timestamp(), + datetime(2013, 10, 11, 23, 59, 59, tzinfo=timezone.utc).timestamp()) + assert chart.x_range == period + assert chart.points[-1][0] < (period[0] + period[1]) / 2 + 3600 + + (output_dir / "1121419604.json").write_text(json.dumps(create_result())) + screen._poll() + await _settle(app, pilot) + + assert screen.query_one("#summary").border_title == "2026-10-08_16-00-57 (local): Completed" + assert screen._poll_timer is None + + _run_app(test) + + +def test_project_keys_do_not_run_actions_on_other_screens() -> None: + async def test(app: LeanApp, pilot) -> None: + create_local_backtest(Path.cwd() / "CSharp Project", "2026-10-08_16-00-57", create_result()) + await pilot.press("v") + await _settle(app, pilot) + + with mock.patch.object(ui_app, "CommandProcess") as command_process: + await pilot.press("b") + assert isinstance(app.screen, BacktestScreen) + command_process.assert_not_called() + + _run_app(test) + + +def test_projects_can_update_while_another_screen_is_shown() -> None: + async def test(app: LeanApp, pilot) -> None: + await pilot.press("p") + assert isinstance(app.screen, ConfirmScreen) + + app._set_projects(app._projects[:1], logged_in=True, status="updated") + await pilot.pause() + + assert app.screen_stack[0].query_one(ProjectTable).row_count == 1 + + _run_app(test, logged_in=True) + + +async def _open_cloud_orders(app: LeanApp, pilot, backtest: dict, order_list: list) -> BacktestScreen: + """Shows a cloud backtest's orders tab, with the API serving order_list.""" + project = UIProject("Cloud Project", cloud_id=123) + post = mock_cloud_api(backtest, order_list=order_list) + post.side_effect = _only_backtest(post.side_effect) + app.push_screen(BacktestScreen(project, logged_in=True)) + await _settle(app, pilot) + await pilot.press("2") + await _settle(app, pilot) + return app.screen + + +def _only_backtest(post): + """Lists only the backtest under test, so the screen opens on it.""" + def answer(endpoint, payload): + if endpoint == "backtests/list": + return {"backtests": [{"backtestId": BACKTEST_ID, "name": "Logical Red Monkey", + "created": "2026-10-08 19:45:10"}]} + return post(endpoint, payload) + return answer + + +class _OrdersAfter(list): + """Orders that the fake orders endpoint only returns from its request number `after` + 1 on.""" + + def __init__(self, after: int, orders: list) -> None: + super().__init__(orders) + self._after = after + + def __getitem__(self, index): + if isinstance(index, slice): + self._after -= 1 + if self._after >= 0: + return [] + return super().__getitem__(index) + + def __len__(self) -> int: + return super().__len__() if self._after < 0 else 0 + + +def _order_requests() -> list: + return [(c.args[1]["start"], c.args[1]["end"]) for c in container.api_client.post.call_args_list + if c.args[0] == "backtests/orders/read"] + + +def _orders_label(screen: BacktestScreen) -> str: + return str(screen.query_one(TabbedContent).get_tab("orders").label) + + +def test_orders_load_a_page_at_a_time_as_they_are_scrolled_to() -> None: + async def test(app: LeanApp, pilot) -> None: + screen = await _open_cloud_orders(app, pilot, create_cloud_backtest(), create_cloud_orders(250)) + table = screen.query_one("#orders-table") + + assert table.row_count == 100 + assert _orders_label(screen) == "Orders (100 of 250)" + assert _order_requests() == [(0, 100)] + + table.move_cursor(row=95) + await _settle(app, pilot) + assert table.row_count == 200 + + table.move_cursor(row=195) + await _settle(app, pilot) + assert table.row_count == 250 + assert _orders_label(screen) == "Orders (250)" + assert _order_requests() == [(0, 100), (100, 200), (200, 300)] + + # Every order is shown, so reaching the end again asks for nothing + table.move_cursor(row=249) + await _settle(app, pilot) + assert len(_order_requests()) == 3 + + _run_app(test, logged_in=True) + + +def test_orders_of_a_running_backtest_add_new_orders_on_refresh() -> None: + async def test(app: LeanApp, pilot) -> None: + order_list = create_cloud_orders(3) + running = create_cloud_backtest(status="Running", completed=False, progress=0.5) + screen = await _open_cloud_orders(app, pilot, running, order_list) + assert screen.query_one("#orders-table").row_count == 3 + + order_list.extend(create_cloud_orders(2, first_id=4)) + screen._poll() + await _settle(app, pilot) + + assert screen.query_one("#orders-table").row_count == 5 + assert _order_requests() == [(0, 100), (3, 103)] + + _run_app(test, logged_in=True) + + +def test_orders_reload_when_the_backtest_completes() -> None: + async def test(app: LeanApp, pilot) -> None: + running = create_cloud_backtest(status="Running", completed=False, progress=0.5) + screen = await _open_cloud_orders(app, pilot, running, create_cloud_orders(3)) + + running.update(create_cloud_backtest()) + screen._poll() + await _settle(app, pilot) + + assert screen.query_one("#orders-table").row_count == 3 + assert _order_requests()[-1] == (0, 100) + assert _orders_label(screen) == "Orders (3)" + + _run_app(test, logged_in=True) + + +def test_orders_that_are_not_readable_yet_are_retried() -> None: + async def test(app: LeanApp, pilot) -> None: + backtest = create_cloud_backtest() + backtest["statistics"]["Total Orders"] = "2" + + # The endpoint returns no orders the first two times it is asked + order_list = _OrdersAfter(2, create_cloud_orders(2)) + with mock.patch.object(backtest_screen, "ORDERS_RETRY_SECONDS", 0.1): + screen = await _open_cloud_orders(app, pilot, backtest, order_list) + for _ in range(10): + if screen.query_one("#orders-table").row_count: + break + await pilot.pause(0.1) + await _settle(app, pilot) + + assert screen.query_one("#orders-table").row_count == 2 + assert _orders_label(screen) == "Orders (2)" + assert len(_order_requests()) == 3 + + _run_app(test, logged_in=True) + + +def test_orders_of_a_running_cloud_backtest_ignore_the_length_without_orders() -> None: + async def test(app: LeanApp, pilot) -> None: + running = create_cloud_backtest(status="In Progress...", completed=False, progress=0.26) + screen = await _open_cloud_orders(app, pilot, running, []) + serve = container.api_client.post.side_effect + + # The cloud serves a running backtest's orders as an empty list with a length of 100 + container.api_client.post.side_effect = lambda endpoint, payload: \ + {"orders": [], "length": 100, "success": True} if endpoint == "backtests/orders/read" \ + else serve(endpoint, payload) + screen._poll() + await _settle(app, pilot) + + assert screen.query_one("#orders-table").row_count == 0 + assert _orders_label(screen) == "Orders (none yet)" + + _run_app(test, logged_in=True) diff --git a/tests/ui/test_results.py b/tests/ui/test_results.py new file mode 100644 index 00000000..f2185a8f --- /dev/null +++ b/tests/ui/test_results.py @@ -0,0 +1,389 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional +from unittest import mock + +import pytest + +from lean.components.api.backtest_client import BacktestClient +from lean.container import container +from lean.models.api import QCBacktest +from lean.ui.results import (CloudBacktest, LocalBacktest, list_cloud_backtests, list_local_backtests, + parse_cloud_backtest, parse_local_result, parse_order, sort_newest_first) + +# 2013-10-07 04:00 UTC to 2013-10-11 20:00 UTC, like the sample data backtests +START = 1381118400 +END = 1381521600 + +BACKTEST_ID = "c9c84362d35b5ab2e8bd457446b74d58" + + +def create_equity_chart(points: int = 5) -> Dict[str, Any]: + """Returns a Strategy Equity chart, whose equity series holds candles.""" + step = (END - START) / max(1, points - 1) + return {"name": "Strategy Equity", "chartType": 0, "series": { + "Equity": {"name": "Equity", "unit": "$", "index": 0, "seriesType": 2, + "values": [[START + i * step, 100000 + i, 100000 + i + 5, 100000 + i - 5, 100000 + i * 10] + for i in range(points)]}, + "Return": {"name": "Return", "unit": "%", "index": 1, "seriesType": 3, "values": [[START, 0.0]]}, + }} + + +def create_orders() -> Dict[str, Any]: + return { + "2": {"type": 0, "id": 2, "symbol": {"value": "SPY", "id": "SPY R735QTJ8XC9X", "permtick": "SPY"}, + "price": 145.34, "time": "2013-10-07T14:46:00Z", "quantity": -686.0, "status": 3, + "tag": "Liquidated", "direction": 1, "value": -99705.8}, + "1": {"type": 0, "id": 1, "symbol": {"value": "SPY", "id": "SPY R735QTJ8XC9X", "permtick": "SPY"}, + "price": 145.33, "time": "2013-10-07T14:10:00Z", "quantity": 686.0, "status": 3, + "tag": "", "direction": 0, "value": 99693.94}, + } + + +def create_result(status: str = "Completed", equity_points: int = 5) -> Dict[str, Any]: + """Returns local backtest results in the format LEAN writes them, trimmed to a few points. + + LEAN only adds the algorithm configuration once the backtest has finished. + """ + result = { + "charts": {"Strategy Equity": create_equity_chart(equity_points)}, + "orders": create_orders(), + "statistics": {"Total Orders": "2", "Net Profit": "-0.160%", "Sharpe Ratio": "1.028"}, + "runtimeStatistics": {"Equity": "$99,839.57", "Net Profit": "$-294.86", "Return": "-0.16 %"}, + "state": {"Status": status, "RuntimeError": ""}, + } + if status != "Running": + result["algorithmConfiguration"] = {"startDate": "2013-10-07T00:00:00Z", "endDate": "2013-10-11T23:59:59Z", + "parameters": {"delay-ms": "30"}} + return result + + +def create_local_backtest(project_dir: Path, name: str, result: Optional[Dict[str, Any]], + backtest_id: int = 1121419604) -> Path: + """Writes a backtest output directory the way `lean backtest` leaves it.""" + output_dir = project_dir / "backtests" / name + output_dir.mkdir(parents=True, exist_ok=True) + (output_dir / "config").write_text(json.dumps({"id": backtest_id, "container": "lean_cli_abc"})) + if result is not None: + (output_dir / f"{backtest_id}.json").write_text(json.dumps(result)) + (output_dir / f"{backtest_id}-order-events.json").write_text("[]") + (output_dir / "log.txt").write_text( + "2026-10-08T20:01:00Z TRACE:: Engine.Run(): start\n" + "2026-10-08T20:01:01Z TRACE:: AlgorithmManager.Run(): Begin DataStream - Start: 10/7/2013 12:00:00 AM " + "Stop: 10/11/2013 11:59:59 PM Time: 10/4/2013 3:30:00 PM Warmup: True\n") + return output_dir + + +def create_cloud_backtest(status: str = "Completed.", completed: bool = True, progress: float = 1.0) -> Dict[str, Any]: + """Returns a backtest as the backtests/read endpoint returns it, without chart data.""" + return {"backtestId": BACKTEST_ID, "projectId": 123, "name": "Logical Red Monkey", "status": status, + "completed": completed, + "progress": progress, "created": "2026-10-08 19:45:10", "error": None, + "backtestStart": "2013-10-07 00:00:00", "backtestEnd": "2013-10-11 23:59:59", + "charts": {"Strategy Equity": {"name": "Strategy Equity", "series": {}}}, + "statistics": {"Net Profit": "136.589%"}, "runtimeStatistics": {"Equity": "$236,588.97"}, + "parameterSet": {"roc_window": "150"}} + + +def create_cloud_orders(count: int, first_id: int = 1) -> List[Dict[str, Any]]: + return [dict(create_orders()["1"], id=first_id + i) for i in range(count)] + + +def mock_cloud_api(backtest: Dict[str, Any], orders: int = 0, logs: int = 0, + order_list: Optional[List[Dict[str, Any]]] = None) -> mock.Mock: + """Makes the API client answer the backtest endpoints, paging orders and logs like the API. + + Pass order_list to change the orders the API returns while a test runs. + """ + order_list = order_list if order_list is not None else create_cloud_orders(orders) + log_lines = [f"line {i}" for i in range(logs)] + + def get(endpoint: str, parameters: Dict[str, Any]) -> Dict[str, Any]: + assert endpoint == "backtests/read" + return {"backtest": backtest} + + def post(endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]: + if endpoint == "backtests/chart/read": + assert payload["name"] == "Strategy Equity" + return {"chart": create_equity_chart()} + if endpoint == "backtests/orders/read": + assert payload["end"] - payload["start"] <= 100 + return {"orders": order_list[payload["start"]:payload["end"]], "length": len(order_list)} + if endpoint == "backtests/read/log": + assert payload["end"] - payload["start"] <= 200 + return {"logs": log_lines[payload["start"]:payload["end"]], "length": len(log_lines)} + if endpoint == "backtests/list": + return {"backtests": [{"backtestId": "a" * 32, "name": "Older", "created": "2026-01-01 10:00:00"}, + {"backtestId": "b" * 32, "name": "Newer", "created": "2026-10-08 19:45:10"}]} + raise AssertionError(f"Unexpected endpoint {endpoint}") + + container.api_client.get = mock.Mock(side_effect=get) + container.api_client.post = mock.Mock(side_effect=post) + container.api_client.backtests = BacktestClient(container.api_client) + return container.api_client.post + + +def test_parse_local_result_reads_equity_closes() -> None: + result = parse_local_result(create_result()) + + assert result.equity[0] == (START, 100000) + assert result.equity[-1] == (END, 100040) + + +def test_parse_local_result_skips_empty_points_and_reads_xy_points() -> None: + data = create_result() + data["charts"]["Strategy Equity"]["series"]["Equity"]["values"] = [[START, 1.0], [START + 60, None], + {"x": END, "y": 2.0}] + + assert parse_local_result(data).equity == [(START, 1.0), (END, 2.0)] + + +def test_parse_local_result_reads_orders_sorted_by_id_with_readable_enums() -> None: + orders = parse_local_result(create_result()).orders + + assert [o.id for o in orders] == [1, 2] + assert (orders[0].type, orders[0].direction, orders[0].status) == ("Market", "Buy", "Filled") + assert orders[1].direction == "Sell" + assert orders[1].tag == "Liquidated" + assert orders[0].time == datetime(2013, 10, 7, 14, 10, tzinfo=timezone.utc) + + +def test_parse_order_treats_null_fields_as_missing() -> None: + order = parse_order({"id": None, "type": None, "direction": None, "quantity": None, "price": None, "value": None, + "status": None, "symbol": None, "tag": None, "time": None}) + + assert (order.id, order.quantity, order.price, order.value) == (0, 0.0, 0.0, 0.0) + assert (order.type, order.direction, order.symbol, order.tag) == ("Market", "Buy", "", "") + + +def test_parse_local_result_reads_statistics_status_and_configuration() -> None: + result = parse_local_result(create_result()) + + assert result.statistics["Sharpe Ratio"] == "1.028" + assert result.runtime_statistics["Equity"] == "$99,839.57" + assert (result.status, result.finished, result.progress) == ("Completed", True, 1.0) + assert result.parameters == {"delay-ms": "30"} + assert result.start == datetime(2013, 10, 7, tzinfo=timezone.utc) + + +def test_parse_local_result_of_a_running_backtest() -> None: + result = parse_local_result(create_result(status="Running")) + + assert (result.status, result.finished) == ("Running", False) + assert result.start is None + + +def test_local_backtest_progress_uses_the_date_range_from_the_log_while_running() -> None: + data = create_result(status="Running", equity_points=5) + data["charts"]["Strategy Equity"]["series"]["Equity"]["values"] = \ + data["charts"]["Strategy Equity"]["series"]["Equity"]["values"][:3] + output_dir = create_local_backtest(Path.cwd() / "Project", "2026-10-08_16-00-57", data) + + result = LocalBacktest(output_dir).load() + + assert result.start == datetime(2013, 10, 7, tzinfo=timezone.utc) + assert 0.45 < result.progress < 0.55 + + +def test_local_backtest_finds_its_result_file_from_its_config() -> None: + output_dir = create_local_backtest(Path.cwd() / "Project", "2026-10-08_16-00-57", create_result()) + (output_dir / "9999-summary.json").write_text("{}") + + backtest = LocalBacktest(output_dir) + + assert backtest.result_file == output_dir / "1121419604.json" + assert backtest.load().name == "2026-10-08_16-00-57" + assert backtest.load_orders(0, 100) == (backtest.load().orders, 2) + assert [o.id for o in backtest.load_orders(1, 100)[0]] == [2] + assert backtest.created == datetime(2026, 10, 8, 16, 0, 57).astimezone(timezone.utc) + + +def test_local_backtest_finds_its_result_file_without_a_config() -> None: + output_dir = create_local_backtest(Path.cwd() / "Project", "2026-10-08_16-00-57", create_result()) + (output_dir / "config").unlink() + + assert LocalBacktest(output_dir).result_file == output_dir / "1121419604.json" + + +def test_local_backtest_without_results_or_with_a_partially_written_file_loads_nothing() -> None: + output_dir = create_local_backtest(Path.cwd() / "Project", "2026-10-08_16-00-57", create_result()) + backtest = LocalBacktest(output_dir) + + (output_dir / "1121419604.json").write_text('{"charts": {"Strategy Eq') + assert backtest.load() is None + + (output_dir / "1121419604.json").unlink() + assert backtest.load() is None + assert LocalBacktest(Path.cwd() / "Project" / "backtests" / "missing").load() is None + + +def test_local_backtest_reads_its_log() -> None: + output_dir = create_local_backtest(Path.cwd() / "Project", "2026-10-08_16-00-57", create_result()) + + assert LocalBacktest(output_dir).load_log()[0] == "2026-10-08T20:01:00Z TRACE:: Engine.Run(): start" + + +def test_list_local_backtests_returns_newest_first() -> None: + project_dir = Path.cwd() / "Project" + for name in ["2026-10-08_16-00-57", "2026-10-09_09-30-00", "2026-01-01_00-00-00"]: + create_local_backtest(project_dir, name, create_result()) + + names = [b.name for b in list_local_backtests(project_dir)] + + assert names == ["2026-10-09_09-30-00", "2026-10-08_16-00-57", "2026-01-01_00-00-00"] + assert list_local_backtests(Path.cwd() / "Missing") == [] + assert list_local_backtests(None) == [] + + +def test_parse_cloud_backtest() -> None: + result = parse_cloud_backtest(QCBacktest(**create_cloud_backtest()), {"Strategy Equity": create_equity_chart()}) + + assert result.name == "Logical Red Monkey" + assert (result.status, result.finished, result.progress) == ("Completed", True, 1.0) + assert result.statistics == {"Net Profit": "136.589%"} + assert result.parameters == {"roc_window": "150"} + assert result.equity[-1] == (END, 100040) + assert result.start == datetime(2013, 10, 7, tzinfo=timezone.utc) + + +def test_parse_cloud_backtest_accepts_empty_charts_and_dates() -> None: + backtest = dict(create_cloud_backtest(status="In Queue...", completed=False, progress=0.0), + charts=[], backtestStart="", backtestEnd=None) + + result = parse_cloud_backtest(QCBacktest(**backtest), {}) + + assert (result.status, result.start, result.end, result.equity) == ("In Queue", None, None, []) + + +def test_cloud_backtest_without_charts_yet_loads_the_summary() -> None: + post = mock_cloud_api(dict(create_cloud_backtest(completed=False, progress=0.1), charts=[])) + + result = CloudBacktest(123, BACKTEST_ID).load() + + assert result.progress == 0.1 + assert all(c.args[0] != "backtests/chart/read" for c in post.call_args_list) + + +def test_parse_cloud_backtest_is_not_finished_until_its_statistics_arrive() -> None: + backtest = dict(create_cloud_backtest(), statistics=[], runtimeStatistics=None) + + result = parse_cloud_backtest(QCBacktest(**backtest), {}) + + assert result.finished is False + + +def test_parse_cloud_backtest_while_running_uses_reported_progress() -> None: + backtest = create_cloud_backtest(status="Running", completed=False, progress=0.42) + backtest["statistics"] = [] + + result = parse_cloud_backtest(QCBacktest(**backtest), {}) + + assert (result.finished, result.progress) == (False, 0.42) + assert result.statistics == {} + assert result.equity == [] + + +def test_cloud_backtest_loads_the_backtest_and_only_its_equity_chart() -> None: + post = mock_cloud_api(create_cloud_backtest()) + + result = CloudBacktest(123, BACKTEST_ID).load() + + assert result.equity[-1] == (END, 100040) + chart_requests = [c for c in post.call_args_list if c.args[0] == "backtests/chart/read"] + assert len(chart_requests) == 1 + assert chart_requests[0].args[1]["start"] == int(datetime(2013, 10, 7, tzinfo=timezone.utc).timestamp()) + + +def test_cloud_backtest_without_a_chart_yet_loads_the_summary() -> None: + backtest = create_cloud_backtest(status="In Queue...", completed=False, progress=0) + backtest["charts"] = {} + post = mock_cloud_api(backtest) + + result = CloudBacktest(123, BACKTEST_ID).load() + + assert result.status == "In Queue" + assert result.equity == [] + assert all(c.args[0] != "backtests/chart/read" for c in post.call_args_list) + + +def test_cloud_backtest_loads_one_page_of_orders_with_the_total() -> None: + post = mock_cloud_api(create_cloud_backtest(), orders=250) + + orders, total = CloudBacktest(123, BACKTEST_ID).load_orders(100, 200) + + assert [o.id for o in orders] == list(range(101, 201)) + assert total == 250 + assert post.call_count == 1 + + +def test_cloud_backtest_never_asks_for_more_than_a_page_of_orders() -> None: + post = mock_cloud_api(create_cloud_backtest(), orders=250) + + orders, _ = CloudBacktest(123, BACKTEST_ID).load_orders(0, 1000) + + assert len(orders) == 100 + assert post.call_args.args[1]["end"] == 100 + + +def test_cloud_backtest_reads_the_last_log_lines() -> None: + mock_cloud_api(create_cloud_backtest(), logs=2500) + + lines = CloudBacktest(123, BACKTEST_ID).load_log() + + assert len(lines) == 2000 + assert (lines[0], lines[-1]) == ("line 500", "line 2499") + + +def test_list_cloud_backtests_returns_newest_first() -> None: + mock_cloud_api(create_cloud_backtest()) + + backtests = list_cloud_backtests(123) + + assert [b.name for b in backtests] == ["Newer", "Older"] + assert backtests[0].created == datetime(2026, 10, 8, 19, 45, 10, tzinfo=timezone.utc) + + +def test_sort_newest_first_mixes_local_and_cloud_backtests() -> None: + local = LocalBacktest(Path("backtests") / "2026-05-01_12-00-00") + cloud: List = [CloudBacktest(1, "a" * 32, "Old", datetime(2025, 1, 1, tzinfo=timezone.utc)), + CloudBacktest(1, "b" * 32, "New", datetime(2026, 10, 1, tzinfo=timezone.utc))] + + assert [b.name for b in sort_newest_first([local] + cloud)] == ["New", "2026-05-01_12-00-00", "Old"] + + +def test_cloud_backtest_waits_while_the_api_prepares_the_orders() -> None: + post = mock_cloud_api(create_cloud_backtest(), orders=3) + serve = post.side_effect + answers = iter([{"status": "loading", "progress": 0.0, "success": True}] * 2) + post.side_effect = lambda endpoint, payload: next(answers, None) or serve(endpoint, payload) + + with mock.patch.object(BacktestClient, "_LOADING_DELAY_SECONDS", 0): + orders, total = CloudBacktest(123, BACKTEST_ID).load_orders(0, 100) + + assert (len(orders), total) == (3, 3) + assert post.call_count == 3 + + +def test_cloud_backtest_gives_up_when_the_api_keeps_loading() -> None: + container.api_client.post = mock.Mock(return_value={"status": "loading", "progress": 0.5, "success": True}) + container.api_client.backtests = BacktestClient(container.api_client) + + with mock.patch.object(BacktestClient, "_LOADING_DELAY_SECONDS", 0), \ + pytest.raises(RuntimeError, match="still preparing"): + CloudBacktest(123, BACKTEST_ID).load_orders(0, 100) + + assert container.api_client.post.call_count == 30 diff --git a/tests/ui/test_ticks.py b/tests/ui/test_ticks.py new file mode 100644 index 00000000..9b6ac396 --- /dev/null +++ b/tests/ui/test_ticks.py @@ -0,0 +1,77 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from datetime import datetime, timezone + +import pytest + +from lean.ui.ticks import format_value_ticks, time_ticks, value_ticks + + +def _labels(low: float, high: float, max_ticks: int, unit: str = "") -> list: + ticks, step = value_ticks(low, high, max_ticks) + return format_value_ticks(ticks, step, unit) + + +@pytest.mark.parametrize("low,high,max_ticks,unit,expected", [ + (97_980, 100_800, 5, "$", ["$98k", "$99k", "$100k"]), + (100_000, 236_588, 5, "$", ["$100k", "$150k", "$200k"]), + (1.394e6, 6.435e6, 5, "$", ["$2M", "$3M", "$4M", "$5M", "$6M"]), + (-0.84, 0.82, 5, "%", ["-0.5%", "0.0%", "0.5%"]), + (-1.34, 0, 3, "%", ["-1.0%", "-0.5%", "0.0%"]), + (0, 1, 5, "", ["0.00", "0.25", "0.50", "0.75", "1.00"]), +]) +def test_value_ticks_are_round_and_formatted_with_unit(low, high, max_ticks, unit, expected) -> None: + assert _labels(low, high, max_ticks, unit) == expected + + +@pytest.mark.parametrize("low,high,max_ticks", [ + (0.11, 0.14, 2), (0.111, 0.119, 2), (99_839.1, 99_839.6, 3), (-5e-7, 3e-7, 3), (1, 1e9, 3), +]) +def test_value_ticks_are_never_empty_and_stay_within_the_range(low, high, max_ticks) -> None: + ticks, _ = value_ticks(low, high, max_ticks) + + assert 0 < len(ticks) <= max_ticks + assert all(low - 1e-12 <= t <= high + 1e-12 for t in ticks) + + +def test_value_ticks_of_a_flat_series() -> None: + assert value_ticks(5, 5, 4) == ([5], 0.0) + + +def _timestamp(*args) -> float: + return datetime(*args, tzinfo=timezone.utc).timestamp() + + +def test_time_ticks_use_days_for_a_week() -> None: + _, labels = time_ticks(_timestamp(2013, 10, 7, 4), _timestamp(2013, 10, 11, 20), 6) + + assert labels == ["2013-10-08", "2013-10-09", "2013-10-10", "2013-10-11"] + + +def test_time_ticks_use_hours_within_a_day() -> None: + _, labels = time_ticks(_timestamp(2013, 10, 7, 13, 30), _timestamp(2013, 10, 7, 18), 5) + + assert labels == ["10-07 14:00", "10-07 15:00", "10-07 16:00", "10-07 17:00", "10-07 18:00"] + + +def test_time_ticks_use_years_for_decades() -> None: + _, labels = time_ticks(_timestamp(2000, 1, 1), _timestamp(2026, 10, 7), 6) + + assert labels == ["2000", "2005", "2010", "2015", "2020", "2025"] + + +def test_time_ticks_use_months_for_a_year() -> None: + _, labels = time_ticks(_timestamp(2023, 1, 15), _timestamp(2023, 12, 20), 6) + + assert labels == ["2023-03", "2023-05", "2023-07", "2023-09", "2023-11"]