Add static code analysis to CI

.github/workflows/build.yml

@@ -25,6 +25,13 @@ jobs:
           pip install setuptools==69.5.1 wheel
           pip install -r requirements.txt
 
+      # Static analysis tools
+      - name: Static Code Analysis
+        if: runner.os == 'Linux' && matrix.python-version != '3.8'
+        run: |
+          pip install mypy==1.19.1 flake8==7.3.0
+          python static_analysis.py
+
       - name: Run tests
         run: python -m pytest -s -rs
 

lean/commands/lean.py

@@ -10,7 +10,6 @@
 # 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 typing import Optional
 
 from click import group, option, Context, pass_context, echo
 

lean/commands/live/deploy.py

@@ -12,7 +12,7 @@
 # limitations under the License.
 
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Tuple
+from typing import List, Optional, Tuple
 from click import option, argument, Choice
 from lean.click import LeanCommand, PathParameter
 from lean.components.util.name_rename import rename_internal_config_to_user_friendly_format

lean/components/api/live_client.py

@@ -11,11 +11,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from datetime import datetime
 from typing import List, Optional
 
 from lean.components.api.api_client import *
-from lean.models.api import QCFullLiveAlgorithm, QCLiveAlgorithmStatus, QCMinimalLiveAlgorithm, QCNotificationMethod, QCRestResponse
+from lean.models.api import QCFullLiveAlgorithm, QCMinimalLiveAlgorithm, QCNotificationMethod, QCRestResponse
 
 
 class LiveClient:

lean/components/util/compiler.py

@@ -111,13 +111,14 @@ def _compile() -> Dict[str, Any]:
         "mounts": [],
         "volumes": {}
     }
-    lean_runner.mount_project_and_library_directories(project_dir, run_options)
-    lean_runner.setup_language_specific_run_options(run_options, project_dir, algorithm_file, False, False)
 
     project_config = project_config_manager.get_project_config(project_dir)
     engine_image = cli_config_manager.get_engine_image(
         project_config.get("engine-image", None))
 
+    lean_runner.mount_project_and_library_directories(project_dir, run_options)
+    lean_runner.setup_language_specific_run_options(run_options, project_dir, algorithm_file, False, False, engine_image)
+
     message["result"] = docker_manager.run_image(engine_image, **run_options)
     temp_manager.delete_temporary_directories_when_done = False
     return message
@@ -153,8 +154,7 @@ def _parse_python_errors(python_output: str, color_coding_required: bool) -> lis
                 errors.append(f"{bcolors.FAIL}Build Error File: {match[0]} Line {match[1]} Column {match[2]} - {match[3]}{bcolors.ENDC}\n")
             else:
                 errors.append(f"Build Error File: {match[0]} Line {match[1]} Column {match[2]} - {match[3]}\n")
-
-        for match in re.findall(r"\*\*\* Sorry: ([^(]+) \(([^,]+), line (\d+)\)", python_output):
+        for match in findall(r"\*\*\* Sorry: ([^(]+) \(([^,]+), line (\d+)\)", python_output):
             if color_coding_required:
                 errors.append(f"{bcolors.FAIL}Build Error File: {match[1]} Line {match[2]} Column 0 - {match[0]}{bcolors.ENDC}\n")
             else:

lean/components/util/project_manager.py

@@ -366,7 +366,6 @@ def restore_csharp_project(self, csproj_file: Path, no_local: bool) -> None:
         """
         from shutil import which
         from subprocess import run, STDOUT, PIPE
-        from lean.models.errors import MoreInfoError
 
         if no_local:
             return

lean/main.py

@@ -97,7 +97,7 @@ def main() -> None:
         if temp_manager.delete_temporary_directories_when_done:
             temp_manager.delete_temporary_directories()
     except Exception as exception:
-        from traceback import format_exc, print_exc
+        from traceback import format_exc
         from click import UsageError, Abort
         from requests import exceptions
         from io import StringIO

static_analysis.py

@@ -0,0 +1,133 @@
+# 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 subprocess
+import sys
+
+MYPY_IGNORE_PATTERNS = [
+    'click.', 'subprocess.', 'Module "',
+    'has incompatible type "Optional',
+    'validator', 'pydantic', '__call__',
+    'OnlyValueValidator', 'V1Validator',
+    'QCParameter', 'QCBacktest'
+]
+
+def run_mypy_check():
+    result = subprocess.run(
+        ["python", "-m", "mypy", "lean/",
+         "--show-error-codes",
+         "--no-error-summary",
+         "--ignore-missing-imports",
+         "--check-untyped-defs"],
+        capture_output=True,
+        text=True
+    )
+
+    errors = []
+    for line in result.stdout.splitlines() + result.stderr.splitlines():
+        if not line.strip() or '[call-arg]' not in line:
+            continue
+
+        # Skip false positives
+        if any(pattern in line for pattern in MYPY_IGNORE_PATTERNS):
+            continue
+
+        errors.append(line.strip())
+
+    return errors
+
+def run_flake8_check(select_code):
+    result = subprocess.run(
+        ["python", "-m", "flake8", "lean/",
+         f"--select={select_code}",
+         "--ignore=ALL",
+         "--exit-zero"],
+        capture_output=True,
+        text=True
+    )
+
+    errors = [line.strip() for line in result.stdout.splitlines() if line.strip()]
+    return errors, len(errors)
+
+def display_errors(title, errors, is_critical = True):
+    level = "CRITICAL" if is_critical else "WARNING"
+    if errors:
+        print(f"{level}: {len(errors)} {title} found:")
+        for error in errors:
+            # Clean path for better display
+            clean_error = error.replace('/home/runner/work/lean-cli/lean-cli/', '')
+            print(f"  {clean_error}")
+    else:
+        print(f"No {title} found")
+
+def display_warning_summary(unused_count):
+    print("\nWarnings:")
+    if unused_count > 0:
+        print(f"  - Unused imports: {unused_count}")
+    print("  Consider addressing warnings in future updates.")
+
+def run_analysis() -> int:
+    print("Running static analysis...")
+    print("=" * 60)
+
+    critical_error_count = 0
+    warning_count = 0
+
+    # Check for missing function arguments with mypy
+    print("\n1. Checking for missing function arguments...")
+    print("-" * 40)
+
+    call_arg_errors = run_mypy_check()
+    display_errors("function call argument mismatch(es)", call_arg_errors)
+    critical_error_count += len(call_arg_errors)
+
+    # Check for undefined variables with flake8
+    print("\n2. Checking for undefined variables...")
+    print("-" * 40)
+
+    undefined_errors, undefined_count = run_flake8_check("F821")
+    display_errors("undefined variable(s)", undefined_errors)
+    critical_error_count += undefined_count
+
+    # Check for unused imports with flake8
+    print("\n3. Checking for unused imports...")
+    print("-" * 40)
+
+    unused_imports, unused_count = run_flake8_check("F401")
+    display_errors("unused import(s)", unused_imports, is_critical=False)
+    warning_count += unused_count
+
+    # Summary
+    print("\n" + "=" * 60)
+
+    if critical_error_count > 0:
+        print(f"BUILD FAILED: Found {critical_error_count} critical error(s)")
+        print("\nSummary of critical errors:")
+        print(f"  - Function call argument mismatches: {len(call_arg_errors)}")
+        print(f"  - Undefined variables: {undefined_count}")
+
+        if warning_count > 0:
+            display_warning_summary(unused_count)
+
+        return 1
+
+    if warning_count > 0:
+        print(f"BUILD PASSED with {warning_count} warning(s)")
+        display_warning_summary(unused_count)
+        return 0
+
+    print("SUCCESS: All checks passed with no warnings")
+    return 0
+
+if __name__ == "__main__":
+    sys.exit(run_analysis())