diff --git a/README.md b/README.md index 6b676bcb..577e04b1 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` @@ -2290,6 +2291,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/docs/images/lean-ui-backtest-completed.png b/docs/images/lean-ui-backtest-completed.png new file mode 100644 index 00000000..7d0e3bcb Binary files /dev/null and b/docs/images/lean-ui-backtest-completed.png differ diff --git a/docs/images/lean-ui-backtest-running.png b/docs/images/lean-ui-backtest-running.png new file mode 100644 index 00000000..81ea653c Binary files /dev/null and b/docs/images/lean-ui-backtest-running.png differ diff --git a/lean/__main__.py b/lean/__main__.py new file mode 100644 index 00000000..e206fb37 --- /dev/null +++ b/lean/__main__.py @@ -0,0 +1,25 @@ +# 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 + +import signal + +# `lean ui` starts commands in their own process group on Windows and stops them with Ctrl+Break, +# which is turned into Ctrl+C here so the command's own Ctrl+C handling (e.g. stopping its Docker container) runs +if hasattr(signal, "SIGBREAK"): + signal.signal(signal.SIGBREAK, lambda *_: signal.raise_signal(signal.SIGINT)) + +from lean.main import main + +main(prog_name="lean") 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..8b51cb08 --- /dev/null +++ b/lean/commands/ui.py @@ -0,0 +1,36 @@ +# 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 + + +@command(cls=LeanCommand, requires_lean_config=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/main.py b/lean/main.py index 061c721b..dce9ba37 100644 --- a/lean/main.py +++ b/lean/main.py @@ -88,10 +88,13 @@ 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 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 + """ 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/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..98fce327 --- /dev/null +++ b/lean/ui/actions.py @@ -0,0 +1,135 @@ +# 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 os +import re +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path +from shlex import quote +from typing import Callable, List, Optional + +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), +] + +# `lean cloud backtest` prints the url of the backtest it creates, which holds the project and backtest ids +CLOUD_BACKTEST_URL = re.compile(r"quantconnect\.com/project/(\d+)/([0-9a-f]{32})") + + +def format_command(args: List[str]) -> str: + """Formats arguments as the `lean` command a user would type.""" + 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 + self._process: Optional[subprocess.Popen] = None + self._stop_requested = False + + 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 + """ + env = dict(os.environ, PYTHONIOENCODING="utf-8", PYTHONUNBUFFERED="1", COLUMNS="120") + creation_flags = subprocess.CREATE_NEW_PROCESS_GROUP if sys.platform == "win32" else 0 + + self._process = subprocess.Popen([sys.executable, "-m", "lean"] + self.args, + cwd=self._cwd, + env=env, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + encoding="utf-8", + errors="replace", + creationflags=creation_flags) + + for line in self._process.stdout: + on_line(line.rstrip("\r\n")) + + return self._process.wait() + + def stop(self) -> None: + """Asks the command to stop the way Ctrl+C would, so it can clean up its Docker containers. + + Calling this again while the command is still running kills it. + """ + if self._process is None or self._process.poll() is not None: + return + + if self._stop_requested: + self._process.kill() + return + self._stop_requested = True + + import signal + if sys.platform == "win32": + self._process.send_signal(signal.CTRL_BREAK_EVENT) + else: + self._process.send_signal(signal.SIGINT) diff --git a/lean/ui/app.py b/lean/ui/app.py new file mode 100644 index 00000000..da8a9a9e --- /dev/null +++ b/lean/ui/app.py @@ -0,0 +1,396 @@ +# 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 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.widgets import DataTable, Footer, Header, Input, OptionList, RichLog, Static +from textual.widgets.option_list import Option + +from lean.ui.actions import ACTIONS, CLOUD_BACKTEST_URL, CommandProcess, ProjectAction, format_command +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 + + +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]): + """Shows the command an action is about to run and asks for confirmation.""" + + BINDINGS = [ + Binding("y,enter", "confirm", "Run"), + Binding("n,escape", "cancel", "Cancel"), + ] + + def __init__(self, command: str) -> None: + super().__init__() + self._command = command + + def compose(self) -> ComposeResult: + message = Text.assemble(("Run this command?\n\n", "bold"), (self._command, "cyan"), + ("\n\n", ""), ("y", "bold"), (" / enter to run, ", "dim"), + ("n", "bold"), (" / esc to cancel", "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 + + 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" + 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" + + table.focus() + self.action_refresh() + + # Loading projects + + def action_refresh(self) -> None: + 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 + self._query(ProjectTable).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) + def _row_highlighted(self) -> None: + self._show_project(self.selected_project) + + @on(DataTable.RowSelected) + 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) + + for action in ACTIONS: + reason = action.unavailable_reason(project, self._logged_in) + 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, a in enumerate(ACTIONS) if a.unavailable_reason(project, self._logged_in) 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) + 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("%Y-%m-%d_%H-%M-%S") + 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) + + @work(thread=True, group="command") + def _run_command(self, action: ProjectAction, args: List[str]) -> None: + process = CommandProcess(args, self._root) + self._process = process + self.call_from_thread(self._command_started, args) + + try: + exit_code = process.run(lambda line: self.call_from_thread(self._command_output, line)) + except Exception as exception: + self.call_from_thread(self._command_output, f"Could not run the command: {exception}") + exit_code = -1 + + self._process = None + self.call_from_thread(self._command_finished, exit_code, action.refreshes_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" + + 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) + match = CLOUD_BACKTEST_URL.search(line) + if match is not None and isinstance(live_screen.live_source, PendingCloudBacktest): + live_screen.set_live_source(CloudBacktest(int(match.group(1)), match.group(2))) + + def _command_finished(self, exit_code: int, refresh_projects: bool) -> 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: + if self._process is None: + return + self._process.stop() + self._query("#output", RichLog).border_subtitle = "stopping, x again to force" + + async def action_quit(self) -> None: + if self._process is not None: + self._process.stop() + self.exit() diff --git a/lean/ui/backtest_screen.py b/lean/ui/backtest_screen.py new file mode 100644 index 00000000..7cbd54cf --- /dev/null +++ b/lean/ui/backtest_screen.py @@ -0,0 +1,702 @@ +# 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.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 textual_plotext import PlotextPlot + +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(PlotextPlot): + """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. + """ + + 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._replot() + + def on_resize(self) -> None: + self._replot() + + def _replot(self) -> None: + plt = self.plt + plt.clear_figure() + + if not self._points: + plt.title(self._message) + self.refresh() + return + + xs = [x for x, _ in self._points] + ys = [y for _, y in self._points] + plt.plot(xs, ys, color="cyan", marker="braille") + + y_ticks, y_step = value_ticks(min(ys), max(ys), max(3, min(8, self.size.height // 2))) + plt.yticks(y_ticks, format_value_ticks(y_ticks, y_step, "$")) + x_low, x_high = self.x_range + if x_high > x_low: + plt.xlim(x_low, x_high) + x_ticks, x_labels = time_ticks(x_low, x_high, max(2, self.size.width // 16)) + plt.xticks(x_ticks, x_labels) + + self.refresh() + + +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 + was_selected = self._is_live(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 was_selected: + self._selected = 0 + self._result = None + self._update_position() + self._load() + else: + self._selected = next((i for i, b in enumerate(self._backtests) if b is self.backtest), 0) + + def add_live_line(self, line: str) -> None: + if self._live is None: + return + self._live.lines.append(line) + if 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_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: + self.notify(f"Could not load the backtest: {error}", severity="error") + 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..f5d3c002 --- /dev/null +++ b/lean/ui/projects.py @@ -0,0 +1,147 @@ +# 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 os import walk +from pathlib import Path +from typing import List, Optional + +from lean.constants import PROJECT_CONFIG_FILE_NAME +from lean.models.api import QCProject + +LANGUAGE_LABELS = {"Python": "Py", "CSharp": "C#", "Py": "Py", "C#": "C#"} + + +@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 = [] + + 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) + projects.append(UIProject( + name=current_path.relative_to(root).as_posix(), + path=current_path, + cloud_id=config.get("cloud-id"), + language=LANGUAGE_LABELS.get(config.get("algorithm-language", ""), ""), + 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=LANGUAGE_LABELS.get(cloud_project.language.value, 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 + + cli_config_manager = container.cli_config_manager + return cli_config_manager.user_id.get_value() is not None and cli_config_manager.api_token.get_value() is not None + + +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..40c0c957 --- /dev/null +++ b/lean/ui/results.py @@ -0,0 +1,440 @@ +# 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 json import loads +from re import compile +from time import sleep +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +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 + +# How often and how long to wait for the cloud to prepare a backtest's chart, orders or logs +CLOUD_LOADING_ATTEMPTS = 30 +CLOUD_LOADING_DELAY_SECONDS = 1 + +# 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 = compile(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, (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: + symbol = data.get("symbol") or {} + order_type = data.get("type", 0) + direction = data.get("direction", 0) + return Order(id=data.get("id", 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", 0.0), + price=data.get("price", 0.0), + value=data.get("value", 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: Dict[str, Any], charts: Dict[str, Any]) -> BacktestResult: + """Parses a backtest from the backtests/read endpoint and its equity chart from backtests/chart/read.""" + statistics = backtest.get("statistics") + parameters = backtest.get("parameterSet") + return BacktestResult(name=backtest.get("name", ""), + equity=parse_equity(charts), + statistics=statistics if isinstance(statistics, dict) else {}, + runtime_statistics=backtest.get("runtimeStatistics") or {}, + status=(backtest.get("status") or "").rstrip("."), + finished=bool(backtest.get("completed")), + runtime_error=backtest.get("error") or "", + start=_parse_time(backtest.get("backtestStart")), + end=_parse_time(backtest.get("backtestEnd")), + parameters=parameters if isinstance(parameters, dict) else {}, + reported_progress=backtest.get("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, "%Y-%m-%d_%H-%M-%S").astimezone(timezone.utc) + except ValueError: + return None + + @property + def result_file(self) -> Optional[Path]: + """The result file, which LEAN names after the backtest id.""" + config_file = self.path / "config" + if config_file.is_file(): + try: + result_file = self.path / f"{loads(config_file.read_text(encoding='utf-8'))['id']}.json" + if result_file.is_file(): + return result_file + except (ValueError, KeyError): + pass + + 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 + try: + result = parse_local_result(loads(result_file.read_text(encoding="utf-8"))) + except ValueError: + 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.""" + 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 = _DATA_STREAM_RANGE.search(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 + + def _post(self, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]: + """Makes a request about this backtest. + + 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, so those answers are retried. + """ + from lean.container import container + + for attempt in range(CLOUD_LOADING_ATTEMPTS): + response = container.api_client.post(endpoint, {"projectId": self.project_id, + "backtestId": self.backtest_id, **payload}) + if response.get("status") != "loading": + return response + if attempt < CLOUD_LOADING_ATTEMPTS - 1: + sleep(CLOUD_LOADING_DELAY_SECONDS) + raise RuntimeError(f"QuantConnect is still preparing the data of {endpoint}, try again with r") + + def load(self) -> Optional[BacktestResult]: + backtest = self._post("backtests/read", {})["backtest"] + self._name = backtest.get("name") or self._name + self._created = _parse_time(backtest.get("created")) or self._created + + charts = {} + start, end = _parse_time(backtest.get("backtestStart")), _parse_time(backtest.get("backtestEnd")) + if EQUITY_CHART in (backtest.get("charts") or {}) and start is not None and end is not None: + try: + chart = self._post("backtests/chart/read", {"name": EQUITY_CHART, "count": CLOUD_CHART_POINTS, + "start": int(start.timestamp()), + "end": int(end.timestamp())}).get("chart") + 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._post("backtests/orders/read", {"start": start, "end": min(end, start + ORDERS_PAGE)}) + return parse_orders(page.get("orders") or []), page.get("length", 0) + + def load_log(self) -> List[str]: + first = self._post("backtests/read/log", {"start": 0, "end": CLOUD_LOG_PAGE, "query": " "}) + 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 = self._post("backtests/read/log", {"start": position, "end": min(position + CLOUD_LOG_PAGE, length), + "query": " "}) + 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 + + data = container.api_client.post("backtests/list", {"projectId": project_id, "includeStatistics": False}) + backtests = [CloudBacktest(project_id, b["backtestId"], b.get("name", ""), _parse_time(b.get("created"))) + for b in data.get("backtests") or []] + 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/setup.py b/setup.py index 8eae6950..649e8425 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", + "textual-plotext>=1.0.0", ] setup( diff --git a/tests/commands/test_ui.py b/tests/commands/test_ui.py new file mode 100644 index 00000000..e1b9a618 --- /dev/null +++ b/tests/commands/test_ui.py @@ -0,0 +1,598 @@ +# 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 +from datetime import datetime, timezone +from pathlib import Path +from unittest import mock + +from click.testing import CliRunner +from textual.widgets import TabbedContent + +from lean.commands import lean +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, CLOUD_BACKTEST_URL, format_command +from lean.ui.app import ConfirmScreen, LeanApp, ProjectTable +from lean.ui.backtest_screen import BacktestPicker, BacktestScreen +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 +from tests.test_ui_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_cloud_backtest_url_holds_the_project_and_backtest_ids() -> None: + match = CLOUD_BACKTEST_URL.search(f"Backtest url: https://www.quantconnect.com/project/19213997/{BACKTEST_ID}") + + assert match.groups() == ("19213997", BACKTEST_ID) + + +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) + + +def test_command_process_runs_the_cli_as_a_child_process(fake_filesystem) -> None: + from lean import __version__ + from lean.ui.actions import CommandProcess + + lines = [] + fake_filesystem.pause() + try: + exit_code = CommandProcess(["--version"], Path(__file__).parent).run(lines.append) + finally: + fake_filesystem.resume() + + assert exit_code == 0 + assert lines == [f"lean {__version__}"] + + +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() + + +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_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_backtesting_locally_opens_the_results_while_the_backtest_runs() -> None: + async def test(app: LeanApp, pilot) -> None: + def fake_run(on_line): + output_dir = Path(command_process.call_args.args[0][3]) + on_line("Starting LEAN") + 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(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 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_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/test_ui_results.py b/tests/test_ui_results.py new file mode 100644 index 00000000..b0bd8487 --- /dev/null +++ b/tests/test_ui_results.py @@ -0,0 +1,347 @@ +# 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.container import container +from lean.ui.results import (CloudBacktest, LocalBacktest, list_cloud_backtests, list_local_backtests, + parse_cloud_backtest, parse_local_result, 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, "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 post(endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]: + if endpoint == "backtests/read": + return {"backtest": backtest} + 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.post = mock.Mock(side_effect=post) + 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_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(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_while_running_uses_reported_progress() -> None: + backtest = create_cloud_backtest(status="Running", completed=False, progress=0.42) + backtest["statistics"] = [] + + result = parse_cloud_backtest(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("lean.ui.results.CLOUD_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}) + + with mock.patch("lean.ui.results.CLOUD_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/test_ui_ticks.py b/tests/test_ui_ticks.py new file mode 100644 index 00000000..9b6ac396 --- /dev/null +++ b/tests/test_ui_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"]