diff --git a/.github/workflows/alpaca-live-broker.yml b/.github/workflows/alpaca-live-broker.yml new file mode 100644 index 000000000..0106292d8 --- /dev/null +++ b/.github/workflows/alpaca-live-broker.yml @@ -0,0 +1,116 @@ +name: Alpaca Live Broker + +on: + workflow_dispatch: + push: + branches: + - dev + paths: + - "lumibot/**" + - "tests/test_alpaca_live_broker_apitest.py" + - "tests/test_agent_runtime_remote_mcp.py" + - "tests/test_agent_runtime_mcp_transports.py" + - "tests/test_agent_runtime_provider_keys.py" + - "tests/test_agent_runtime_errors.py" + - "tests/test_agent_tool_permissions.py" + - "tests/test_agent_alpaca_news_builtin.py" + - "tests/test_agent_alpaca_news_live_apitest.py" + - "tests/backtest/test_agent_runtime_backtest.py" + - "tests/backtest/test_ai_committee_builtin_tools_backtest.py" + - "requirements*.txt" + - "pyproject.toml" + - "setup.py" + - ".github/workflows/alpaca-live-broker.yml" + pull_request: + branches: [dev, main] + paths: + - "lumibot/**" + - "tests/test_alpaca_live_broker_apitest.py" + - "tests/test_agent_runtime_remote_mcp.py" + - "tests/test_agent_runtime_mcp_transports.py" + - "tests/test_agent_runtime_provider_keys.py" + - "tests/test_agent_runtime_errors.py" + - "tests/test_agent_tool_permissions.py" + - "tests/test_agent_alpaca_news_builtin.py" + - "tests/test_agent_alpaca_news_live_apitest.py" + - "tests/backtest/test_agent_runtime_backtest.py" + - "tests/backtest/test_ai_committee_builtin_tools_backtest.py" + - "requirements*.txt" + - "pyproject.toml" + - "setup.py" + - ".github/workflows/alpaca-live-broker.yml" + +permissions: + contents: read + +env: + AIOHTTP_NO_EXTENSIONS: 1 + BACKTESTING_DATA_SOURCE: none + BACKTESTING_SHOW_PROGRESS_BAR: "false" + ALPACA_TEST_API_KEY: ${{ secrets.ALPACA_TEST_API_KEY }} + ALPACA_TEST_API_SECRET: ${{ secrets.ALPACA_TEST_API_SECRET }} + ALPACA_NEWS_API_KEY: ${{ secrets.ALPACA_NEWS_API_KEY || secrets.ALPACA_TEST_API_KEY }} + ALPACA_NEWS_API_SECRET: ${{ secrets.ALPACA_NEWS_API_SECRET || secrets.ALPACA_TEST_API_SECRET }} + +jobs: + alpaca-live-broker: + name: Alpaca live broker strategy + runs-on: blacksmith-2vcpu-ubuntu-2404 + timeout-minutes: 55 + if: >- + github.actor != 'dependabot[bot]' && + (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) + concurrency: + group: paper-broker-live-tests-${{ github.repository }} + cancel-in-progress: false + + steps: + - uses: actions/checkout@v4 + with: + persist-credentials: false + + - name: Set up Python 3.10 + uses: actions/setup-python@v5 + with: + python-version: "3.10" + cache: pip + + - name: Install dependencies + run: | + set -euo pipefail + echo "Set AIOHTTP_NO_EXTENSIONS=$AIOHTTP_NO_EXTENSIONS so that aiohttp doesn't try to install C extensions" + python -m pip install --upgrade pip + pip install requests + pip install -r requirements_dev.txt + + - name: Verify Alpaca paper credentials are available + run: | + set -euo pipefail + if [ -z "${ALPACA_TEST_API_KEY}" ] || [ -z "${ALPACA_TEST_API_SECRET}" ]; then + echo "::error::ALPACA_TEST_API_KEY and ALPACA_TEST_API_SECRET repository secrets are required for Alpaca live broker CI." + exit 1 + fi + if [ -z "${ALPACA_NEWS_API_KEY}" ] || [ -z "${ALPACA_NEWS_API_SECRET}" ]; then + echo "::error::ALPACA_NEWS_API_KEY and ALPACA_NEWS_API_SECRET, or fallback ALPACA_TEST_API_KEY and ALPACA_TEST_API_SECRET, are required for Alpaca news live apitests." + exit 1 + fi + + - name: Run agent MCP and built-in action coverage + run: | + set -euo pipefail + timeout 1500 python -m pytest -q \ + tests/test_agent_runtime_remote_mcp.py \ + tests/test_agent_runtime_mcp_transports.py \ + tests/test_agent_runtime_provider_keys.py \ + tests/test_agent_runtime_errors.py \ + tests/test_agent_tool_permissions.py \ + tests/test_agent_alpaca_news_builtin.py \ + tests/test_agent_alpaca_news_live_apitest.py \ + tests/backtest/test_agent_runtime_backtest.py \ + tests/backtest/test_ai_committee_builtin_tools_backtest.py + + - name: Run real Alpaca broker strategy apitests + run: | + set -euo pipefail + timeout 1200 python -m pytest -q -m apitest --tb=short \ + tests/test_alpaca_live_broker_apitest.py diff --git a/.github/workflows/cicd.yaml b/.github/workflows/cicd.yaml index 8f2a8b68a..a6a949d1b 100644 --- a/.github/workflows/cicd.yaml +++ b/.github/workflows/cicd.yaml @@ -3,13 +3,14 @@ name: LumiBot CI/CD on: # Full LumiBot CI is intentionally expensive: lint + 6 unit shards + 4 backtest - # shards. Run it when explicitly requested, for release/CI tags, or for main PRs. + # shards. Run it when explicitly requested, for release/CI tags, or for PRs to + # either long-lived branch. push: tags: - "v*.*.*" - "ci-*" pull_request: - branches: [main] + branches: [main, dev] paths: - "lumibot/**" - "tests/**" @@ -54,10 +55,12 @@ jobs: timeout-minutes: 15 environment: unit-tests steps: - - uses: actions/checkout@v3 + - uses: actions/checkout@v4 + with: + persist-credentials: false - name: Set up Python 3.10 - uses: actions/setup-python@v4 + uses: actions/setup-python@v5 with: python-version: "3.10" cache: pip @@ -99,10 +102,12 @@ jobs: shard: [0, 1, 2, 3, 4, 5] steps: - - uses: actions/checkout@v3 + - uses: actions/checkout@v4 + with: + persist-credentials: false - name: Set up Python 3.10 - uses: actions/setup-python@v4 + uses: actions/setup-python@v5 with: python-version: "3.10" cache: pip @@ -123,7 +128,7 @@ jobs: # CI target: keep PR checks fast and deterministic. # Downloader + apitests are run separately / opt-in. - PYTEST_MARKERS='not apitest and not downloader' + PYTEST_MARKERS='not apitest and not downloader and not broker_strategy_live' export PYTEST_MARKERS echo "Shard ${SHARD_INDEX}/${SHARD_TOTAL} markers=${PYTEST_MARKERS}" @@ -174,8 +179,10 @@ jobs: ) PY - echo "Running $(wc -l shard_files.txt | awk '{print $1}') files" - timeout 1500 python -m pytest -m "${PYTEST_MARKERS}" --tb=short -q --durations=30 -x $(cat shard_files.txt) + mapfile -t shard_files < shard_files.txt + ((${#shard_files[@]} > 0)) || { echo "No unit-test files selected"; exit 1; } + echo "Running ${#shard_files[@]} files" + timeout 1500 python -m pytest -m "${PYTEST_MARKERS}" --tb=short -q --durations=30 -x "${shard_files[@]}" backtest-tests: name: Backtest Tests (shard ${{ matrix.shard }}/4) @@ -189,10 +196,12 @@ jobs: shard: [0, 1, 2, 3] steps: - - uses: actions/checkout@v3 + - uses: actions/checkout@v4 + with: + persist-credentials: false - name: Set up Python 3.10 - uses: actions/setup-python@v4 + uses: actions/setup-python@v5 with: python-version: "3.10" cache: pip @@ -223,7 +232,7 @@ jobs: run: | set -euo pipefail - PYTEST_MARKERS='not apitest and not downloader' + PYTEST_MARKERS='not apitest and not downloader and not broker_strategy_live' export PYTEST_MARKERS echo "Shard ${SHARD_INDEX}/${SHARD_TOTAL} markers=${PYTEST_MARKERS}" @@ -264,27 +273,104 @@ jobs: print(f"selected_nodeids={len(selected)} total_nodeids={len(nodeids)}") PY - echo "Running $(wc -l shard_nodeids.txt | awk '{print $1}') nodeids" - timeout 1500 python -m pytest -m "${PYTEST_MARKERS}" --tb=short -q --durations=30 -x $(cat shard_nodeids.txt) + mapfile -t shard_nodeids < shard_nodeids.txt + ((${#shard_nodeids[@]} > 0)) || { echo "No backtest nodeids selected"; exit 1; } + echo "Running ${#shard_nodeids[@]} nodeids" + timeout 1500 python -m pytest -m "${PYTEST_MARKERS}" --tb=short -q --durations=30 -x "${shard_nodeids[@]}" + + broker-live-strategy-tests: + name: Broker Live Strategy Tests + runs-on: ubuntu-latest + timeout-minutes: 20 + environment: unit-tests + needs: lint + if: >- + github.event_name != 'pull_request' || + (github.event.pull_request.head.repo.full_name == github.repository && github.actor != 'dependabot[bot]') + concurrency: + group: paper-broker-live-tests-${{ github.repository }} + cancel-in-progress: false + + steps: + - uses: actions/checkout@v4 + with: + persist-credentials: false + + - name: Set up Python 3.10 + uses: actions/setup-python@v5 + with: + python-version: "3.10" + cache: pip + + - name: Install dependencies + run: | + echo "Set AIOHTTP_NO_EXTENSIONS=$AIOHTTP_NO_EXTENSIONS so that aiohttp doesn't try to install C extensions" + python -m pip install --upgrade pip + pip install requests + pip install -r requirements_dev.txt + + - name: Validate broker paper secrets + run: | + set -euo pipefail + python - <<'PY' + import os + + required = [ + "ALPACA_TEST_API_KEY", + "ALPACA_TEST_API_SECRET", + "TRADIER_TEST_ACCESS_TOKEN", + "TRADIER_TEST_ACCOUNT_NUMBER", + ] + missing = [name for name in required if not os.environ.get(name)] + if missing: + raise SystemExit("Missing broker live strategy test secrets: " + ", ".join(missing)) + + print("Broker live strategy test secrets present.") + PY + + - name: Run broker live strategy tests + run: | + set -euo pipefail + timeout 900 python -m pytest -q --tb=short -x \ + -m "broker_strategy_live" \ + tests/test_broker_live_strategy_run_apitest.py LintAndTest: name: LintAndTest runs-on: ubuntu-latest if: always() - needs: [lint, unit-tests, backtest-tests] + needs: [lint, unit-tests, backtest-tests, broker-live-strategy-tests] steps: - name: Check results + env: + ACTOR: ${{ github.actor }} + BACKTEST_TESTS_RESULT: ${{ needs.backtest-tests.result }} + BROKER_LIVE_RESULT: ${{ needs.broker-live-strategy-tests.result }} + EVENT_NAME: ${{ github.event_name }} + LINT_RESULT: ${{ needs.lint.result }} + PR_HEAD_REPOSITORY: ${{ github.event.pull_request.head.repo.full_name }} + REPOSITORY: ${{ github.repository }} + UNIT_TESTS_RESULT: ${{ needs.unit-tests.result }} run: | - echo "lint: ${{ needs.lint.result }}" - echo "unit-tests: ${{ needs.unit-tests.result }}" - echo "backtest-tests: ${{ needs.backtest-tests.result }}" + echo "lint: $LINT_RESULT" + echo "unit-tests: $UNIT_TESTS_RESULT" + echo "backtest-tests: $BACKTEST_TESTS_RESULT" + echo "broker-live-strategy-tests: $BROKER_LIVE_RESULT" - if [ "${{ needs.lint.result }}" != "success" ]; then + if [ "$LINT_RESULT" != "success" ]; then + exit 1 + fi + if [ "$UNIT_TESTS_RESULT" != "success" ]; then exit 1 fi - if [ "${{ needs.unit-tests.result }}" != "success" ]; then + if [ "$BACKTEST_TESTS_RESULT" != "success" ]; then exit 1 fi - if [ "${{ needs.backtest-tests.result }}" != "success" ]; then + if [ "$BROKER_LIVE_RESULT" = "skipped" ] && \ + [ "$EVENT_NAME" = "pull_request" ] && \ + { [ "$PR_HEAD_REPOSITORY" != "$REPOSITORY" ] || \ + [ "$ACTOR" = "dependabot[bot]" ]; }; then + echo "broker-live-strategy-tests: intentionally skipped for untrusted pull request" + elif [ "$BROKER_LIVE_RESULT" != "success" ]; then exit 1 fi diff --git a/docs/ALPACA_LIVE_BROKER_CI.md b/docs/ALPACA_LIVE_BROKER_CI.md new file mode 100644 index 000000000..5992de4cf --- /dev/null +++ b/docs/ALPACA_LIVE_BROKER_CI.md @@ -0,0 +1,70 @@ +# Alpaca Live Broker CI + +One-line description: Real Alpaca paper-account CI for live broker and agent-tool smoke coverage. + +Last Updated: 2026-07-03 + +Status: Active + +Audience: Developers, AI Agents + +## Overview + +`.github/workflows/alpaca-live-broker.yml` runs opt-in live paper API coverage for Alpaca. It is separate from the normal `not apitest` suite because it contacts Alpaca, reads broker/data endpoints, and submits then cancels real paper orders. + +The workflow runs for internal PRs/pushes and manual dispatches when relevant broker, agent, dependency, or workflow files change. Fork PRs are skipped because repository secrets are unavailable. + +## Credentials + +Repository secrets required: + +- `ALPACA_TEST_API_KEY` +- `ALPACA_TEST_API_SECRET` + +Optional repository secrets: + +- `ALPACA_NEWS_API_KEY` +- `ALPACA_NEWS_API_SECRET` + +The live news apitest uses `ALPACA_NEWS_API_KEY` / `ALPACA_NEWS_API_SECRET` when present and falls back to `ALPACA_TEST_API_KEY` / `ALPACA_TEST_API_SECRET` for CI/local smoke runs. The built-in `alpaca_news` tool intentionally does not read generic `ALPACA_API_KEY` values unless it is bound to an active Alpaca broker. Do not commit real key values. + +## Test Coverage + +`tests/test_alpaca_live_broker_apitest.py` is marked `apitest` and `alpaca`. + +- `_require_alpaca()` validates paper credentials, authenticates through the real Alpaca API, fails if the paper account is trading-blocked, and configures the broker test market as `24/7` so scheduled one-shot tests run outside NYSE hours. +- `_LiveOrderDataStrategy` calls `run_live(run_once=True)`, reads AAPL last price and daily bars, submits a non-marketable AAPL limit buy, cancels it, waits for terminal cancel state, and verifies order retrieval. +- `_LiveOptionsChainStrategy` calls `run_live(run_once=True)`, reads SPY price, pulls SPY option chains through the broker data source, selects an expiration, and resolves a valid call contract. + +`tests/test_agent_alpaca_news_live_apitest.py` uses Alpaca news credentials, falling back to the paper test credentials, to verify the built-in Alpaca news agent tool against real historical news pagination and full-content reads. + +The workflow also runs local-only agent runtime, MCP transport, permission, provider-key, and backtest tool coverage before the live broker apitests. + +## Concurrency + +The live broker job has a repository-wide concurrency group. This prevents overlapping runs from using the same shared Alpaca paper account at once and avoids one run canceling or observing another run's test orders. + +## Local Run + +Use paper credentials only: + +```bash +export ALPACA_TEST_API_KEY="..." +export ALPACA_TEST_API_SECRET="..." + +python -m pytest -q \ + tests/test_agent_runtime_remote_mcp.py \ + tests/test_agent_runtime_mcp_transports.py \ + tests/test_agent_runtime_provider_keys.py \ + tests/test_agent_runtime_errors.py \ + tests/test_agent_tool_permissions.py \ + tests/test_agent_alpaca_news_builtin.py \ + tests/test_agent_alpaca_news_live_apitest.py \ + tests/backtest/test_agent_runtime_backtest.py \ + tests/backtest/test_ai_committee_builtin_tools_backtest.py + +python -m pytest -q -m apitest --tb=short \ + tests/test_alpaca_live_broker_apitest.py +``` + +If credentials are missing, `alpaca`-marked apitests skip with the missing Alpaca variables instead of requiring unrelated Polygon or ThetaData credentials. diff --git a/docs/SMART_LIMIT_LIVE_TESTING.md b/docs/SMART_LIMIT_LIVE_TESTING.md index d30bc2fee..c0854cc7d 100644 --- a/docs/SMART_LIMIT_LIVE_TESTING.md +++ b/docs/SMART_LIMIT_LIVE_TESTING.md @@ -45,7 +45,29 @@ Notes: - Tradier smoke includes a paper connectivity check; the submit/cancel lifecycle test requires explicit live config (`TRADIER_IS_PAPER=false`) and should be run only when you intend to touch the live API. -### 4) Run live matrix (market hours) +### 4) Strict CI strategy-run smoke + +CI runs a strict paper-broker strategy path for Alpaca and Tradier: + +- instantiates the real broker from `ALPACA_TEST_*` / `TRADIER_TEST_*` secrets; +- submits, polls, and cancels a direct broker GTC limit order as a broker contract preflight; +- runs `Trader.run_all(run_once=True)`; +- executes `before_starting_trading`, `on_trading_iteration`, and `on_strategy_end`; +- submits a deep non-marketable AAPL GTC limit order through `Strategy.submit_order(...)`; +- polls the real paper broker for the order id/status; +- cancels through `Strategy.cancel_order(...)`; +- fails if required secrets are missing or the broker order does not reach canceled state. + +The strategy-run test overrides only the executor's market-open gate so pull-request CI is not tied to US equity +session timing. Broker config and submitted assets stay unchanged; the paper broker still accepts or rejects the real +GTC order. + +```bash +/Users/robertgrzesik/bin/safe-timeout 1200s python3 -m pytest -q -m "broker_strategy_live" \ + tests/test_broker_live_strategy_run_apitest.py +``` + +### 5) Run live matrix (market hours) These are heavier tests that add: @@ -60,7 +82,7 @@ These are heavier tests that add: tests/test_smart_limit_live_matrix_tradier.py ``` -### 5) Benchmarks (market hours; not a pytest) +### 6) Benchmarks (market hours; not a pytest) Benchmarks are in `scripts/` and write CSVs to `logs/`. These are used for price-improvement statistics and are not treated as strict pass/fail (paper fills can be unrealistic). diff --git a/docsrc/deployment.rst b/docsrc/deployment.rst index 59879df49..191a90dc1 100644 --- a/docsrc/deployment.rst +++ b/docsrc/deployment.rst @@ -405,7 +405,7 @@ Coinbase is a cryptocurrency broker that is easy to set up and operates across a - organizations/a7df3e75-5gg5-4b0d-805c-e91c02fd63b8/apiKeys/1abb999e-8442-4607-lkc7-423eb8d478e3 * - COINBASE_PRIVATE_KEY - Your private key for Coinbase. **Required** if you are using Coinbase as your broker. - - -----BEGIN EC PRIVATE KEY-----\nPLjCAQEEIFOxfolkj7JmTkEUyctOqAq0hQt02SRBy7GnJHGQyb56jToAoGCCqGSM49\nAwEHoUQDQgAEg1VBKEVkqhy+9eHxeao7b7cMsbXXeB/Ggm2sYKEm2Ebrhq67Nobj\n5ze8ddf78UFICjOcooHovd+1oFcZZ+RLQ==\n-----END EC PRIVATE KEY-----\n" + - * - COINBASE_API_PASSPHRASE - Your API passphrase for Coinbase. **Optional** if you are using Coinbase as your broker. - 123456 diff --git a/docsrc/environment_variables.rst b/docsrc/environment_variables.rst index 1d757dc5a..b758539b7 100644 --- a/docsrc/environment_variables.rst +++ b/docsrc/environment_variables.rst @@ -388,6 +388,16 @@ ALPACA_IS_PAPER - Values: ``true`` (paper) / ``false`` (live). - Default: ``true`` (paper trading). +ALPACA_TEST_API_KEY / ALPACA_TEST_API_SECRET +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +- Purpose: Alpaca paper credentials used by ``pytest.mark.alpaca`` API tests and the Alpaca live-broker CI workflow. +- Values: Alpaca paper API credentials (**do not hardcode**). +- Notes: + - These tests may submit and cancel non-marketable paper orders. + - Live news API tests use ``ALPACA_NEWS_API_KEY`` / ``ALPACA_NEWS_API_SECRET`` when present and fall back to these paper test credentials for CI/local smoke runs. + - Missing values skip ``alpaca``-marked API tests; invalid or trading-blocked paper accounts fail the live-broker checks. + ALPACA_NEWS_API_KEY / ALPACA_NEWS_API_SECRET ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ diff --git a/lumibot/brokers/alpaca.py b/lumibot/brokers/alpaca.py index ba4d96896..2251abff9 100644 --- a/lumibot/brokers/alpaca.py +++ b/lumibot/brokers/alpaca.py @@ -964,7 +964,16 @@ def _flatten_order(self, order): - def _submit_orders(self, orders, is_multileg=False, order_type=None, duration="day", price=None): + def _submit_orders( + self, + orders, + is_multileg=False, + order_type=None, + duration="day", + price=None, + take_profit=None, + stop_loss=None, + ): """ Submit multiple orders to the broker. Supports multi-leg (MLeg) orders for options. """ @@ -973,7 +982,15 @@ def _submit_orders(self, orders, is_multileg=False, order_type=None, duration="d if is_multileg: tag = orders[0].tag if hasattr(orders[0], "tag") and orders[0].tag else orders[0].strategy - parent_order = self._submit_multileg_order(orders, order_type, duration, price, tag) + parent_order = self._submit_multileg_order( + orders, + order_type, + duration, + price, + tag, + take_profit=take_profit, + stop_loss=stop_loss, + ) return [parent_order] else: sub_orders = [] @@ -981,7 +998,16 @@ def _submit_orders(self, orders, is_multileg=False, order_type=None, duration="d sub_orders.append(self._submit_order(order)) return sub_orders - def _submit_multileg_order(self, orders, order_type="limit", duration="day", price=None, tag=None): + def _submit_multileg_order( + self, + orders, + order_type="limit", + duration="day", + price=None, + tag=None, + take_profit=None, + stop_loss=None, + ): """ Submit a multi-leg (MLeg) options order to Alpaca. @@ -989,7 +1015,7 @@ def _submit_multileg_order(self, orders, order_type="limit", duration="day", pri - Tradier uses "credit" for net credit (receive premium) and "debit" for net debit (pay premium). - Alpaca only supports "market" and "limit" for multi-leg orders. - We convert "credit", "debit", and "even" to "limit" for Alpaca, as both are limit orders in Alpaca's API. - - The sign of the limit price (positive/negative) is not used by Alpaca to distinguish credit/debit. + - Bracket exits use Alpaca's take_profit and stop_loss fields on the same mleg request. - Alpaca requires that the leg ratio quantities are relatively prime (GCD == 1). """ requested_multileg_type = order_type if order_type in ("credit", "debit", "even") else None @@ -1087,7 +1113,15 @@ def _submit_multileg_order(self, orders, order_type="limit", duration="day", pri if price is not None: # Ensure limit price is at most 2 decimal places (Alpaca requirement) limit_price = round(float(price), 2) + if requested_multileg_type == "credit": + limit_price = -abs(limit_price) + elif requested_multileg_type == "debit": + limit_price = abs(limit_price) kwargs["limit_price"] = limit_price + if take_profit is not None: + kwargs["take_profit"] = self._format_multileg_take_profit(take_profit) + if stop_loss is not None: + kwargs["stop_loss"] = self._format_multileg_stop_loss(stop_loss) # Submit order try: response = self.api.submit_order(order_data=OrderData(**kwargs)) @@ -1116,6 +1150,32 @@ def _submit_multileg_order(self, orders, order_type="limit", duration="day", pri o.set_error(e) raise + @staticmethod + def _format_multileg_take_profit(take_profit): + if not isinstance(take_profit, dict): + raise ValueError("multi-leg take_profit must be a dict") + limit_price = take_profit.get("limit_price") + if limit_price is None: + raise ValueError("multi-leg take_profit requires limit_price") + return {"limit_price": round(float(limit_price), 2)} + + @staticmethod + def _format_multileg_stop_loss(stop_loss): + if not isinstance(stop_loss, dict): + raise ValueError("multi-leg stop_loss must be a dict") + unsupported = {"trail_price", "trail_percent"} & set(stop_loss) + if unsupported: + names = ", ".join(sorted(unsupported)) + raise ValueError(f"multi-leg stop_loss does not support {names}") + payload = {} + if stop_loss.get("stop_price") is not None: + payload["stop_price"] = round(float(stop_loss["stop_price"]), 2) + if stop_loss.get("limit_price") is not None: + payload["limit_price"] = round(float(stop_loss["limit_price"]), 2) + if "stop_price" not in payload: + raise ValueError("multi-leg stop_loss requires stop_price") + return payload + def _submit_order(self, order): """Submit an order for an asset (single-leg, including options)""" diff --git a/lumibot/brokers/tradovate.py b/lumibot/brokers/tradovate.py index 1c39e2324..85e8b45f5 100644 --- a/lumibot/brokers/tradovate.py +++ b/lumibot/brokers/tradovate.py @@ -116,7 +116,8 @@ def __init__(self, config=None, data_source=None, connect_stream=None): self._validate_config() if connect_stream is None: - connect_stream = self._connect_on_init + connect_stream = True + self._connect_stream_requested = bool(connect_stream) if data_source is None: config["TRADING_API_URL"] = self.trading_api_url @@ -127,7 +128,9 @@ def __init__(self, config=None, data_source=None, connect_stream=None): market_token=self.market_token ) - super().__init__(name=self.NAME, data_source=data_source, config=config, connect_stream=connect_stream) + # Authenticate before launching the polling stream. This keeps construction lazy while + # preserving polling for normal live brokers after their first authenticated operation. + super().__init__(name=self.NAME, data_source=data_source, config=config, connect_stream=False) if self._connect_on_init: self._ensure_connected() @@ -160,6 +163,7 @@ def _ensure_connected(self): with self._tradovate_connection_lock: if self._tradovate_connected: + self._ensure_polling_stream() return self._tradovate_connecting = True @@ -183,6 +187,7 @@ def _ensure_connected(self): self.user_id = self._get_user_info(self.trading_token) logger.info(colored(f"User ID: {self.user_id}", "green")) self._tradovate_connected = True + self._ensure_polling_stream() except TradovateAPIError as e: logger.warning(colored(f"Failed initial connection to Tradovate: {e}", "yellow")) logger.warning(colored("Broker connection failed due to rate limiting or authentication.", "yellow")) @@ -190,6 +195,13 @@ def _ensure_connected(self): finally: self._tradovate_connecting = False + def _ensure_polling_stream(self): + if not getattr(self, "_connect_stream_requested", False) or getattr(self, "stream", None) is not None: + return + self.stream = self._get_stream_object() + if self.stream is not None: + self._launch_stream() + def _throttle_rest(self): """Ensure REST calls respect a soft per-minute cap.""" if self._rate_limit_per_minute <= 0: diff --git a/lumibot/entities/data.py b/lumibot/entities/data.py index 079c5dbe9..d8d1215d0 100644 --- a/lumibot/entities/data.py +++ b/lumibot/entities/data.py @@ -1256,6 +1256,204 @@ def _get_bars_between_dates_dict(self, timestep=None, start_date=None, end_date= def _validate_bars_request(self, dt, length=1, timestep=None, timeshift=0): return True + def _validate_native_bars_request( + self, + dt, + *, + length=1, + timeshift=0, + request_timestep=None, + stale_request_timestep=_MISSING, + ) -> tuple[int, int]: + """Validate a native-bar request without the generic decorator hot-path cost. + + This preserves ``check_data``'s boundaries and diagnostics while reusing the row + lookup that the native slice already needs. Native option histories may be sparse + inside those boundaries, so their source-level refresh logic owns internal coverage + decisions; treating every quiet option minute as stale causes repeated downloads and + changes established backtest results. ``request_timestep`` governs end-boundary lag; + ``stale_request_timestep`` separately governs bar age inside the frame. Quantity-one + callers pass ``None`` for the latter to retain the default 3-minute non-crypto and + 15-minute crypto tolerances. + """ + if type(length) not in [int, float]: + raise TypeError(f"Length must be an integer. {type(length)} was provided.") + if timeshift is not None and not isinstance(timeshift, (int, float, datetime.timedelta)): + raise TypeError( + f"Timeshift must be a number or datetime.timedelta. {type(timeshift)} was provided." + ) + + dt_key = dt.to_pydatetime() if isinstance(dt, pd.Timestamp) else dt + normalized_timeshift = self._normalize_timeshift_to_rows(timeshift) + if stale_request_timestep is _MISSING: + stale_request_timestep = request_timestep + + if dt_key < self.datetime_start: + raise ValueError( + f"The date you are looking for ({dt_key}) for ({self.asset}) is outside of the data's date range ({self.datetime_start} to {self.datetime_end}). This could be because the data for this asset does not exist for the date you are looking for, or something else." + ) + + if self.timestep == "day" or dt_key > self.datetime_end: + self._validate_native_bars_request_end( + dt_key=dt_key, + length=length, + normalized_timeshift=normalized_timeshift, + request_timestep=request_timestep, + ) + + iter_count = self.get_iter_count(dt_key) + + stale_bar_error = None + allow_sparse_history = getattr(self.asset, "asset_type", None) == Asset.AssetType.OPTION + strict_end_check = getattr(self, "strict_end_check", False) + tolerance_ns = None + if strict_end_check and not allow_sparse_history: + tolerance_cache = self.__dict__.get("_strict_intraday_tolerance_ns_cache") + tolerance_ns = ( + _MISSING if tolerance_cache is None else tolerance_cache.get(stale_request_timestep, _MISSING) + ) + if tolerance_ns is _MISSING: + tolerance = self._strict_intraday_bar_age_tolerance(request_timestep=stale_request_timestep) + tolerance_ns = None if tolerance is None else int(tolerance.total_seconds() * 1_000_000_000) + if tolerance_cache is None: + tolerance_cache = {} + self._strict_intraday_tolerance_ns_cache = tolerance_cache + tolerance_cache[stale_request_timestep] = tolerance_ns + + if strict_end_check and not allow_sparse_history and tolerance_ns is not None: + index_values_ns = getattr(self, "_index_values_ns", None) + if index_values_ns is None: + stale_bar_error = self._strict_intraday_stale_bar_error( + dt_key=dt_key, + iter_count=int(iter_count), + length=length, + timeshift=normalized_timeshift, + request_timestep=stale_request_timestep, + ) + else: + try: + if self.datetime_start.tzinfo is None: + epoch = datetime.datetime(1970, 1, 1) + delta = dt_key - epoch + dt_ns = ( + (delta.days * 86_400 + delta.seconds) * 1_000_000_000 + + delta.microseconds * 1_000 + ) + else: + dt_ns = int(dt_key.timestamp() * 1_000_000_000) + gap_ns = dt_ns - int(index_values_ns[int(iter_count)]) + except (AttributeError, IndexError, OverflowError, TypeError, ValueError): + stale_bar_error = self._strict_intraday_stale_bar_error( + dt_key=dt_key, + iter_count=int(iter_count), + length=length, + timeshift=normalized_timeshift, + request_timestep=stale_request_timestep, + ) + else: + if gap_ns > tolerance_ns: + stale_bar_error = self._strict_intraday_stale_bar_error( + dt_key=dt_key, + iter_count=int(iter_count), + length=length, + timeshift=normalized_timeshift, + request_timestep=stale_request_timestep, + ) + + if stale_bar_error is not None: + raise ValueError(stale_bar_error) + + data_index = int(iter_count) + 1 - length - normalized_timeshift + if data_index < 0: + self._log_native_insufficient_history( + dt_key=dt_key, + length=length, + normalized_timeshift=normalized_timeshift, + iter_count=iter_count, + ) + + return normalized_timeshift, int(iter_count) + + def _validate_native_bars_request_end( + self, + *, + dt_key, + length, + normalized_timeshift: int, + request_timestep, + ) -> None: + """Handle uncommon native-bar end-boundary checks.""" + if self.timestep == "day": + import pytz + + utc = pytz.UTC + if hasattr(self.datetime_end, "astimezone"): + datetime_end_utc = self.datetime_end.astimezone(utc) + else: + datetime_end_utc = self.datetime_end + dt_exceeds_end = dt_key.date() > datetime_end_utc.date() + else: + dt_exceeds_end = dt_key > self.datetime_end + + if dt_exceeds_end: + if getattr(self, "strict_end_check", False): + strict_lag_tolerance = self._strict_end_lag_tolerance(request_timestep=request_timestep) + gap = dt_key - self.datetime_end + if ( + strict_lag_tolerance is None + or gap < datetime.timedelta(0) + or gap > strict_lag_tolerance + ): + raise ValueError( + f"The date you are looking for ({dt_key}) for ({self.asset}) is after the available data's end ({self.datetime_end}) with length={length} and timeshift={normalized_timeshift}; data refresh required instead of using stale bars." + ) + + gap = dt_key - self.datetime_end + max_gap = datetime.timedelta(days=3) + if gap > max_gap: + raise ValueError( + f"The date you are looking for ({dt_key}) for ({self.asset}) is after the available data's end ({self.datetime_end}) with length={length} and timeshift={normalized_timeshift}; data refresh required instead of using stale bars." + ) + message = ( + f"The date you are looking for ({dt_key}) is after the available data's end " + f"({self.datetime_end}) by {gap}. Using the last available bar (within tolerance " + f"of {max_gap})." + ) + if self.timestep == "day": + logger.debug(message) + else: + logger.warning(message) + + def _log_native_insufficient_history( + self, + *, + dt_key, + length, + normalized_timeshift: int, + iter_count: int, + ) -> None: + """Match ``check_data`` diagnostics for a native request before available history.""" + data_index = int(iter_count) + 1 - length - normalized_timeshift + logger.warning( + f"The date you are looking for ({dt_key}) is outside of the data's date range ({self.datetime_start} to {self.datetime_end}) after accounting for a length of {length} and a timeshift of {normalized_timeshift}. Keep in mind that the length you are requesting must also be available in your data, in this case we are {data_index} rows away from the data you need." + ) + try: + idx_vals = self.df.index + logger.info( + "[DATA][CHECK] asset=%s timestep=%s dt=%s length=%s timeshift=%s iter_index=%s idx_min=%s idx_max=%s rows=%s", + getattr(self.asset, "symbol", self.asset), + getattr(self, "timestep", None), + dt_key, + length, + normalized_timeshift, + iter_count, + idx_vals.min(), + idx_vals.max(), + len(idx_vals), + ) + except Exception: + logger.debug("[DATA][CHECK] failed to log index diagnostics", exc_info=True) + def _normalize_timeshift_to_rows(self, timeshift): if timeshift is None: return 0 @@ -1269,10 +1467,11 @@ def _normalize_timeshift_to_rows(self, timeshift): return int(timeshift or 0) - def _get_bars_row_bounds(self, dt, length=1, timeshift=0): + def _get_bars_row_bounds(self, dt, length=1, timeshift=0, iter_count=None): timeshift = self._normalize_timeshift_to_rows(timeshift) - iter_count = self.get_iter_count(dt) + if iter_count is None: + iter_count = self.get_iter_count(dt) try: if pd.isna(iter_count): iter_count = 0 @@ -1313,11 +1512,12 @@ def _get_bars_source_frame(self): return df_source - def _get_bars_frame_window(self, dt, length=1, timeshift=0): + def _get_bars_frame_window(self, dt, length=1, timeshift=0, iter_count=None): start_row, end_row, normalized_timeshift = self._get_bars_row_bounds( dt, length=length, timeshift=timeshift, + iter_count=iter_count, ) df_source = self._get_bars_source_frame() df = df_source._slice(slice(start_row, end_row)) @@ -1396,10 +1596,17 @@ def get_bars(self, dt, length=1, timestep=MIN_TIMESTEP, timeshift=0): and int(native_qty) == int(quantity) and native_unit == "minute" ): + timeshift, iter_count = self._validate_native_bars_request( + dt, + length=num_periods, + timeshift=timeshift, + request_timestep=f"{int(quantity)}{timestep}", + ) df, start_row, end_row, timeshift = self._get_bars_frame_window( dt, length=num_periods, timeshift=timeshift, + iter_count=iter_count, ) if df is None: return None @@ -1486,10 +1693,22 @@ def get_bars(self, dt, length=1, timestep=MIN_TIMESTEP, timeshift=0): or (timestep == "day" and self.timestep == "day") ) ): + timeshift, iter_count = self._validate_native_bars_request( + dt, + length=length, + timeshift=timeshift, + request_timestep=f"{int(quantity)}{timestep}", + stale_request_timestep=None, + ) # PERF: avoid reconstructing a DataFrame from datalines on every call. # The underlying `self.df` is already indexed by datetime, so we can slice by # row bounds in O(1) and return a stable OHLCV schema. - df, start_row, end_row, timeshift = self._get_bars_frame_window(dt, length=length, timeshift=timeshift) + df, start_row, end_row, timeshift = self._get_bars_frame_window( + dt, + length=length, + timeshift=timeshift, + iter_count=iter_count, + ) if df is None: return None diff --git a/requirements.txt b/requirements.txt index 0b7fac61a..c0df5894f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,10 +1,8 @@ polygon-api-client>=1.13.3 alpaca-py>=0.42.0 -alpha_vantage ibapi==9.81.1.post1 yfinance>=0.2.61 matplotlib>=3.3.3 -quandl numpy>=1.20.0,<2.5.0 pandas>=2.2.0 polars>=1.32.3 @@ -12,8 +10,6 @@ pandas_market_calendars>=5.1.0 pandas-ta-classic>=0.3.14b0 plotly>=5.18.0 sqlalchemy -bcrypt -pytest yappi>=1.6.0 scipy>=1.14.0 quantstats-lumi>=1.1.5,<1.2.0 @@ -30,12 +26,9 @@ py-clob-client-v2>=1.0.1 websockets>=15.0.1 pytz psycopg2-binary -exchange_calendars>=4.6.0 duckdb tabulate databento>=0.42.0 -holidays -psutil openai setuptools<81 google-adk[extensions]>=2.0.0,<3.0.0 @@ -44,7 +37,6 @@ litellm>=1.83.7,<=1.83.14 anyio>=4.10.0 mcp>=1.26.0 schwab-py>=1.5.0 -Flask>=2.3 free-proxy requests-oauthlib boto3>=1.40.64 diff --git a/setup.cfg b/setup.cfg index 85ca07953..ef41d0dae 100644 --- a/setup.cfg +++ b/setup.cfg @@ -22,6 +22,7 @@ known-third-party=lumibot [tool:pytest] markers = apitest: marks tests as API tests (deselect with '-m "not apitest"') + broker_strategy_live: strict paper-broker strategy-run tests that submit, poll, and cancel real orders smartlimit_matrix: larger SMART_LIMIT live matrix tests (run with '-m smartlimit_matrix') downloader: marks tests that require the shared Theta downloader service polymarket: marks tests that require Polymarket CLOB credentials or live market data @@ -44,7 +45,7 @@ norecursedirs = docs .* *.egg* appdir jupyter *pycache* venv* .cache* .coverage* # .coveragerc to control coverag.py [coverage:run] -command_line = -m pytest -m "not apitest" +command_line = -m pytest -m "not apitest and not broker_strategy_live" branch = True omit = # * so you can get all dirs diff --git a/setup.py b/setup.py index a53f14fd9..0e53807e2 100644 --- a/setup.py +++ b/setup.py @@ -56,11 +56,9 @@ def _maybe_copy_theta_terminal(self): install_requires=[ "polygon-api-client>=1.13.3", "alpaca-py>=0.42.0", - "alpha_vantage", "ibapi==9.81.1.post1", "yfinance>=0.2.61", "matplotlib>=3.3.3", - "quandl", # NumPy 2.5 emits noisy timedelta deprecation warnings through pandas 2.x calendar paths. "numpy>=1.20.0,<2.5.0", "pandas>=2.2.0", @@ -69,8 +67,6 @@ def _maybe_copy_theta_terminal(self): "pandas-ta-classic>=0.3.14b0", "plotly>=5.18.0", "sqlalchemy", - "bcrypt", - "pytest", "yappi>=1.6.0", # SciPy 1.14.0+ supports NumPy 2.x "scipy>=1.14.0", @@ -88,13 +84,9 @@ def _maybe_copy_theta_terminal(self): "py-clob-client-v2>=1.0.1", "pytz", "psycopg2-binary", - # Exchange calendars 4.6.0+ supports NumPy 2.x - "exchange_calendars>=4.6.0", "duckdb", "tabulate", "databento>=0.42.0", - "holidays", - "psutil", "openai", "setuptools<81", "google-adk[extensions]>=2.0.0,<3.0.0", @@ -103,7 +95,6 @@ def _maybe_copy_theta_terminal(self): "anyio>=4.10.0", "mcp>=1.26.0", "schwab-py>=1.5.0", - "Flask>=2.3", "free-proxy", "requests-oauthlib", "boto3>=1.40.64", diff --git a/tests/conftest.py b/tests/conftest.py index c203df119..55547f16a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,17 +3,17 @@ Includes global cleanup for APScheduler instances to prevent CI hangs. """ -import pytest -import gc import atexit -import threading -import os +import gc import json -from unittest import mock -from pathlib import Path -from dotenv import load_dotenv +import os +import threading from collections import defaultdict +from pathlib import Path +from unittest import mock +import pytest +from dotenv import load_dotenv _TEST_DURATIONS_SECONDS = defaultdict(float) @@ -85,10 +85,26 @@ def _restore_is_backtesting_env(): def pytest_configure(config): - config.addinivalue_line("markers", "ibkr: downloader-only IBKR tests that do not require Polygon or ThetaData credentials") - config.addinivalue_line("markers", "polymarket: Polymarket CLOB tests that do not require Polygon or ThetaData credentials") - config.addinivalue_line("markers", "polymarket_credentials: Polymarket CLOB tests that require authenticated credentials") - config.addinivalue_line("markers", "polymarket_live_trading: Polymarket CLOB tests that can submit/cancel live orders") + config.addinivalue_line( + "markers", + "ibkr: downloader-only IBKR tests that do not require Polygon or ThetaData credentials", + ) + config.addinivalue_line( + "markers", + "alpaca: Alpaca paper API tests that require ALPACA_TEST_API_KEY / ALPACA_TEST_API_SECRET", + ) + config.addinivalue_line( + "markers", + "polymarket: Polymarket CLOB tests that do not require Polygon or ThetaData credentials", + ) + config.addinivalue_line( + "markers", + "polymarket_credentials: Polymarket CLOB tests that require authenticated credentials", + ) + config.addinivalue_line( + "markers", + "polymarket_live_trading: Polymarket CLOB tests that can submit/cancel live orders", + ) class _PatchProxy: @@ -137,7 +153,7 @@ def cleanup_all_schedulers(): try: # Force garbage collection to trigger __del__ methods gc.collect() - + # Try to find and shutdown any remaining APScheduler instances for obj in gc.get_objects(): if hasattr(obj, '__class__') and 'scheduler' in str(obj.__class__).lower(): @@ -159,7 +175,7 @@ def cleanup_all_threads(): # Get all active threads active_threads = threading.enumerate() main_thread = threading.main_thread() - + for thread in active_threads: if thread != main_thread and thread.is_alive(): # Try to stop threads that have a stop method or event @@ -225,17 +241,17 @@ def pytest_sessionfinish(session, exitstatus): @pytest.fixture(scope="session", autouse=True) def global_cleanup(): """Global cleanup fixture that runs at session start and end""" - + # Cleanup before tests start cleanup_all_schedulers() cleanup_all_threads() - + yield - + # Cleanup after all tests complete cleanup_all_schedulers() cleanup_all_threads() - + # Force final garbage collection gc.collect() @@ -244,7 +260,7 @@ def global_cleanup(): def test_cleanup(): """Per-test cleanup to prevent scheduler leaks between tests""" yield - + # Minimal cleanup to avoid CI deadlocks # Only force gc collection, don't do aggressive scheduler cleanup per-test gc.collect() @@ -324,12 +340,14 @@ def pytest_runtest_setup(item: pytest.Item): - polygon: requires Polygon credentials - thetadata: requires ThetaData credentials - ibkr: downloader-only IBKR tests; does not require Polygon/ThetaData creds + - alpaca: Alpaca paper API tests; does not require Polygon/ThetaData creds - polymarket: Polymarket CLOB tests; does not require Polygon/ThetaData creds - polymarket_credentials: requires Polymarket wallet/CLOB credentials - polymarket_live_trading: requires explicit live-trading enablement Behavior: - If a test is marked with ibkr, require neither Polygon nor ThetaData. + - If a test is marked with alpaca, require Alpaca paper credentials only. - If a test is marked with polygon and/or thetadata, only those provider credentials are required. - If a test has apitest/downloader but no provider-specific markers, @@ -344,12 +362,13 @@ def pytest_runtest_setup(item: pytest.Item): requires_polygon = item.get_closest_marker("polygon") is not None requires_theta = item.get_closest_marker("thetadata") is not None requires_ibkr = item.get_closest_marker("ibkr") is not None + requires_alpaca = item.get_closest_marker("alpaca") is not None requires_polymarket = item.get_closest_marker("polymarket") is not None requires_polymarket_credentials = item.get_closest_marker("polymarket_credentials") is not None requires_polymarket_live_trading = item.get_closest_marker("polymarket_live_trading") is not None # Determine which providers are required - if requires_ibkr or requires_polymarket: + if requires_ibkr or requires_alpaca or requires_polymarket: need_polygon = False need_theta = False elif requires_polygon or requires_theta: @@ -376,6 +395,12 @@ def pytest_runtest_setup(item: pytest.Item): if _is_placeholder(theta_pass): missing.append("THETADATA_PASSWORD") + if requires_alpaca: + if _is_placeholder(os.environ.get("ALPACA_TEST_API_KEY")): + missing.append("ALPACA_TEST_API_KEY") + if _is_placeholder(os.environ.get("ALPACA_TEST_API_SECRET")): + missing.append("ALPACA_TEST_API_SECRET") + if requires_polymarket_credentials: if _is_placeholder(os.environ.get("POLYMARKET_PRIVATE_KEY")): missing.append("POLYMARKET_PRIVATE_KEY") diff --git a/tests/test_agent_alpaca_news_live_apitest.py b/tests/test_agent_alpaca_news_live_apitest.py index 1be1a57ad..5bdbb656d 100644 --- a/tests/test_agent_alpaca_news_live_apitest.py +++ b/tests/test_agent_alpaca_news_live_apitest.py @@ -5,8 +5,7 @@ from lumibot.components.agents import BuiltinTools - -pytestmark = pytest.mark.apitest +pytestmark = [pytest.mark.apitest, pytest.mark.alpaca] class _LiveNewsStrategy: @@ -23,21 +22,23 @@ def get_datetime(self): return datetime(2025, 4, 22, 16, 0, tzinfo=timezone.utc) -def _require_alpaca_news_creds() -> None: - has_key = bool(os.environ.get("ALPACA_NEWS_API_KEY") or os.environ.get("ALPACA_API_KEY")) - has_secret = bool(os.environ.get("ALPACA_NEWS_API_SECRET") or os.environ.get("ALPACA_API_SECRET")) - if not (has_key and has_secret): +def _require_alpaca_news_creds(monkeypatch: pytest.MonkeyPatch) -> None: + api_key = os.environ.get("ALPACA_NEWS_API_KEY") or os.environ.get("ALPACA_TEST_API_KEY") + api_secret = os.environ.get("ALPACA_NEWS_API_SECRET") or os.environ.get("ALPACA_TEST_API_SECRET") + if not (api_key and api_secret): pytest.skip("Missing Alpaca news credentials") + monkeypatch.setenv("ALPACA_NEWS_API_KEY", api_key) + monkeypatch.setenv("ALPACA_NEWS_API_SECRET", api_secret) -def test_live_alpaca_news_known_market_event_scan_and_full_content(): +def test_live_alpaca_news_known_market_event_scan_and_full_content(monkeypatch): """Smoke-test real Alpaca/Benzinga historical news quality. The 2024-08-05 market selloff is a known broad-market news day. This verifies the tool retrieves relevant historical articles by symbol/date window, keeps scan mode light, and can fetch full article bodies when explicitly requested. """ - _require_alpaca_news_creds() + _require_alpaca_news_creds(monkeypatch) tool = BuiltinTools.news.alpaca_news().binder(_LiveNewsStrategy(), None) scan = tool.function( @@ -62,7 +63,10 @@ def test_live_alpaca_news_known_market_event_scan_and_full_content(): f"{article.get('headline') or ''} {article.get('summary') or ''}".lower() for article in scan["articles"] ) - assert any(keyword in combined_scan_text for keyword in ("vix", "selloff", "recession", "global", "plunge", "volatility")) + assert any( + keyword in combined_scan_text + for keyword in ("vix", "selloff", "recession", "global", "plunge", "volatility") + ) full = tool.function( symbols="SPY,QQQ,DIA,IWM", @@ -82,12 +86,15 @@ def test_live_alpaca_news_known_market_event_scan_and_full_content(): assert full_content_articles assert max(len(str(article["content"])) for article in full_content_articles) > 1000 assert all(article.get("content_truncated") is False for article in full_content_articles) - assert all(article.get("content_original_length") == len(str(article.get("content") or "")) for article in full_content_articles) + assert all( + article.get("content_original_length") == len(str(article.get("content") or "")) + for article in full_content_articles + ) -def test_live_alpaca_news_pagination_fetches_distinct_second_page(): +def test_live_alpaca_news_pagination_fetches_distinct_second_page(monkeypatch): """Verify real Alpaca pagination, not just unit-level page_token forwarding.""" - _require_alpaca_news_creds() + _require_alpaca_news_creds(monkeypatch) tool = BuiltinTools.news.alpaca_news().binder(_LivePaginationStrategy(), None) first_page = tool.function( diff --git a/tests/test_alpaca_live_broker_apitest.py b/tests/test_alpaca_live_broker_apitest.py new file mode 100644 index 000000000..213df48de --- /dev/null +++ b/tests/test_alpaca_live_broker_apitest.py @@ -0,0 +1,187 @@ +import time +from datetime import datetime, timedelta + +import pytest + +from lumibot.brokers.alpaca import Alpaca +from lumibot.components.options_helper import OptionsHelper +from lumibot.credentials import ALPACA_TEST_CONFIG +from lumibot.entities import Asset, Order +from lumibot.strategies.strategy import Strategy + +pytestmark = [pytest.mark.apitest, pytest.mark.alpaca] + + +def _require_alpaca() -> Alpaca: + api_key = ALPACA_TEST_CONFIG.get("API_KEY") + api_secret = ALPACA_TEST_CONFIG.get("API_SECRET") + if not api_key or not api_secret or api_key == "" or api_secret == "": + pytest.skip("Missing ALPACA_TEST_API_KEY / ALPACA_TEST_API_SECRET") + + config = dict(ALPACA_TEST_CONFIG) + # CI runs outside market hours, but this test must still exercise run_live(run_once=True). + config["MARKET"] = "24/7" + broker = Alpaca(config, max_workers=1, connect_stream=False) + try: + account = broker.api.get_account() + except Exception as exc: + broker.cleanup_streams() + raise RuntimeError(f"Alpaca paper account authentication failed: {exc}") from exc + + if getattr(account, "trading_blocked", False): + broker.cleanup_streams() + pytest.fail("Alpaca paper account is trading-blocked") + + return broker + + +def _status_text(raw_order) -> str: + status = getattr(raw_order, "status", "") + if hasattr(status, "value"): + status = status.value + return str(status).lower() + + +def _wait_for_terminal_cancel(broker: Alpaca, order_id: str, *, timeout: float = 30.0) -> str: + deadline = time.time() + timeout + last_status = "" + while time.time() < deadline: + raw = broker.api.get_order_by_id(order_id) + last_status = _status_text(raw) + if last_status in {"canceled", "cancelled"} or last_status.endswith(".canceled"): + return last_status + if last_status in {"filled", "rejected", "expired"}: + return last_status + time.sleep(0.25) + return last_status + + +class _LiveOrderDataStrategy(Strategy): + def initialize(self): + self.sleeptime = "1S" + self.vars.iterations = 0 + self.vars.submitted_order_id = None + self.vars.stock_price = None + self.vars.bar_count = 0 + + def on_trading_iteration(self): + self.vars.iterations += 1 + + asset = Asset("AAPL", asset_type=Asset.AssetType.STOCK) + price = self.get_last_price(asset) + assert price is not None and float(price) > 0 + self.vars.stock_price = float(price) + + bars = self.get_historical_prices(asset, 3, "day") + assert bars is not None and bars.df is not None and not bars.df.empty + self.vars.bar_count = len(bars.df) + + # Non-marketable by design: CI must prove real submit/cancel wiring without taking a fill. + order = self.create_order( + asset, + 1, + Order.OrderSide.BUY, + order_type=Order.OrderType.LIMIT, + limit_price=0.01, + time_in_force="day", + ) + submitted = self.submit_order(order) + assert submitted is not None and submitted.identifier + self.vars.submitted_order_id = submitted.identifier + self.broker.cancel_order(submitted) + + +class _LiveOptionsChainStrategy(Strategy): + def initialize(self): + self.sleeptime = "1S" + self.options_helper = OptionsHelper(self) + self.vars.iterations = 0 + self.vars.call_expirations = 0 + self.vars.put_expirations = 0 + self.vars.option_symbol = None + + def on_trading_iteration(self): + self.vars.iterations += 1 + + underlying = Asset("SPY", asset_type=Asset.AssetType.STOCK) + underlying_price = self.get_last_price(underlying) + assert underlying_price is not None and float(underlying_price) > 0 + + chains = self.get_chains(underlying) + chain_root = chains.get("Chains", {}) if isinstance(chains, dict) else {} + call_chains = chain_root.get("CALL", {}) + put_chains = chain_root.get("PUT", {}) + assert call_chains, "Alpaca returned no SPY call chains" + assert put_chains, "Alpaca returned no SPY put chains" + self.vars.call_expirations = len(call_chains) + self.vars.put_expirations = len(put_chains) + + target_date = datetime.now().astimezone().date() + timedelta(days=7) + expiry = self.options_helper.get_expiration_on_or_after_date( + target_date, + chains, + "call", + underlying_asset=underlying, + ) + assert expiry is not None + + expiry_key = expiry.strftime("%Y-%m-%d") if hasattr(expiry, "strftime") else str(expiry) + strikes = call_chains.get(expiry_key) + assert strikes, f"Alpaca returned no SPY call strikes for {expiry_key}" + strike = min(strikes, key=lambda value: abs(float(value) - float(underlying_price))) + + option = self.options_helper.find_next_valid_option(underlying, strike, expiry, put_or_call="call") + assert option is not None + self.vars.option_symbol = str(option) + + +def test_alpaca_run_once_strategy_reads_data_submits_and_cancels_real_order(): + broker = _require_alpaca() + strategy = _LiveOrderDataStrategy( + broker=broker, + name="alpaca-live-broker-order-ci", + benchmark_asset=None, + should_backup_variables_to_database=False, + should_send_summary_to_discord=False, + save_logfile=False, + ) + + try: + strategy.run_live(run_once=True) + assert strategy.vars.iterations == 1 + assert strategy.vars.stock_price > 0 + assert strategy.vars.bar_count >= 1 + assert strategy.vars.submitted_order_id + + status = _wait_for_terminal_cancel(broker, strategy.vars.submitted_order_id) + assert status in {"canceled", "cancelled"} or status.endswith(".canceled") + + all_orders = broker._pull_broker_all_orders() + assert all_orders is not None + finally: + try: + strategy.cancel_open_orders() + except Exception: + pass + broker.cleanup_streams() + + +def test_alpaca_run_once_strategy_reads_options_chain_through_broker_data_source(): + broker = _require_alpaca() + strategy = _LiveOptionsChainStrategy( + broker=broker, + name="alpaca-live-broker-options-ci", + benchmark_asset=None, + should_backup_variables_to_database=False, + should_send_summary_to_discord=False, + save_logfile=False, + ) + + try: + strategy.run_live(run_once=True) + assert strategy.vars.iterations == 1 + assert strategy.vars.call_expirations > 0 + assert strategy.vars.put_expirations > 0 + assert strategy.vars.option_symbol + finally: + broker.cleanup_streams() diff --git a/tests/test_alpaca_multileg_fix.py b/tests/test_alpaca_multileg_fix.py index 7ceaa7754..290b21fa3 100644 --- a/tests/test_alpaca_multileg_fix.py +++ b/tests/test_alpaca_multileg_fix.py @@ -102,6 +102,56 @@ def test_multileg_order_class_is_correct(self, mock_trading_client): # Alpaca expects the short code "mleg" for multi-leg orders assert hasattr(order_data, 'order_class'), "OrderData should have order_class attribute" assert order_data.order_class == "mleg", f"Expected order_class 'mleg', got '{order_data.order_class}'" + + @patch('lumibot.brokers.alpaca.TradingClient') + def test_multileg_order_includes_bracket_exits(self, mock_trading_client): + """Test that multi-leg bracket exits are sent on the Alpaca mleg request.""" + mock_trading_client.return_value = Mock() + broker = Alpaca(self.test_config, connect_stream=False) + orders = [ + Order( + strategy="test_strategy", + asset=self.call_asset, + quantity=1, + side="buy_to_open", + order_type=Order.OrderType.MARKET, + ), + Order( + strategy="test_strategy", + asset=Asset( + symbol="SPY", + asset_type=Asset.AssetType.OPTION, + expiration=self.expiration, + strike=455.0, + right="call", + ), + quantity=1, + side="sell_to_open", + order_type=Order.OrderType.MARKET, + ), + ] + + with patch.object(broker, 'api') as mock_api: + mock_response = Mock() + mock_response.id = "test_order_id" + mock_response.status = "submitted" + mock_api.submit_order.return_value = mock_response + + broker._submit_multileg_order( + orders, + order_type="debit", + duration="day", + price=1.25, + tag="spread-bracket-1", + take_profit={"limit_price": 2.0}, + stop_loss={"stop_price": 0.6, "limit_price": 0.55}, + ) + + order_data = mock_api.submit_order.call_args.kwargs["order_data"] + assert order_data.order_class == "mleg" + assert order_data.limit_price == 1.25 + assert order_data.take_profit == {"limit_price": 2.0} + assert order_data.stop_loss == {"stop_price": 0.6, "limit_price": 0.55} @patch('lumibot.brokers.alpaca.TradingClient') def test_multileg_order_has_required_fields(self, mock_trading_client): @@ -223,6 +273,60 @@ def test_multileg_order_with_limit_price(self, mock_trading_client): # This should raise an error because price is None for a limit order broker._submit_multileg_order(orders, order_type="limit", price=None) + @patch('lumibot.brokers.alpaca.TradingClient') + def test_credit_multileg_order_uses_negative_limit_price(self, mock_trading_client): + """Alpaca uses signed mleg limit prices: positive debit, negative credit.""" + mock_trading_client.return_value = Mock() + broker = Alpaca(self.test_config, connect_stream=False) + orders = [ + Order( + strategy="test_strategy", + asset=self.call_asset, + quantity=1, + side="sell_to_open", + order_type=Order.OrderType.MARKET, + ), + Order( + strategy="test_strategy", + asset=Asset( + symbol="SPY", + asset_type=Asset.AssetType.OPTION, + expiration=self.expiration, + strike=455.0, + right="call", + ), + quantity=1, + side="buy_to_open", + order_type=Order.OrderType.MARKET, + ), + ] + + with patch.object(broker, 'api') as mock_api: + mock_response = Mock() + mock_response.id = "test_order_id" + mock_response.status = "submitted" + mock_api.submit_order.return_value = mock_response + + broker._submit_multileg_order(orders, order_type="credit", price=1.25) + + order_data = mock_api.submit_order.call_args.kwargs["order_data"] + assert order_data.side == "sell" + assert order_data.type == "limit" + assert order_data.limit_price == -1.25 + + def test_multileg_stop_loss_rejects_trailing_fields(self): + """Alpaca mleg bracket stop_loss supports stop_price and optional limit_price only.""" + with pytest.raises(ValueError, match="does not support trail_percent"): + Alpaca._format_multileg_stop_loss({"trail_percent": 0.10}) + + with pytest.raises(ValueError, match="requires stop_price"): + Alpaca._format_multileg_stop_loss({"limit_price": 0.55}) + + assert Alpaca._format_multileg_stop_loss({"stop_price": 0.60, "limit_price": 0.55}) == { + "stop_price": 0.60, + "limit_price": 0.55, + } + @patch('lumibot.brokers.alpaca.TradingClient') def test_smart_limit_submits_as_limit(self, mock_trading_client): """Smart limit should downgrade to limit before broker submission.""" diff --git a/tests/test_broker_live_strategy_run_apitest.py b/tests/test_broker_live_strategy_run_apitest.py new file mode 100644 index 000000000..71dad8512 --- /dev/null +++ b/tests/test_broker_live_strategy_run_apitest.py @@ -0,0 +1,321 @@ +import time + +import pytest + +from lumibot.brokers.alpaca import Alpaca +from lumibot.brokers.tradier import Tradier +from lumibot.credentials import ALPACA_TEST_CONFIG, TRADIER_TEST_CONFIG +from lumibot.entities import Asset, Order +from lumibot.strategies.strategy import Strategy +from lumibot.traders.trader import Trader + +pytestmark = [pytest.mark.apitest, pytest.mark.broker_strategy_live] + + +_CANCELLED_STATUSES = {"canceled", "cancelled"} +_TERMINAL_STATUSES = _CANCELLED_STATUSES | {"error", "expired", "fill", "filled", "rejected"} +_LIVE_ACTIVE_STATUSES = { + "accepted", + "accepted_for_bidding", + "held", + "new", + "open", + "partially_filled", + "pending", + "pending_cancel", + "pending_new", + "queued", + "submitted", +} + + +def _normalized_order_status(record): + if record is None: + return None + if isinstance(record, str): + raw_status = record + elif isinstance(record, dict): + raw_status = record.get("status") + else: + raw_status = getattr(record, "status", None) + if hasattr(raw_status, "value"): + raw_status = raw_status.value + if raw_status is None: + return None + status = str(raw_status).lower() + if "." in status: + status = status.rsplit(".", 1)[-1] + return status + + +def _pull_order_status(broker, identifier): + return _normalized_order_status(broker._pull_broker_order(identifier)) + + +def _wait_for_broker_status(broker, identifier, *, timeout=30, expected_statuses=None): + deadline = time.time() + timeout + last_status = None + while time.time() < deadline: + last_status = _pull_order_status(broker, identifier) + if last_status and (expected_statuses is None or last_status in expected_statuses): + return last_status + time.sleep(0.5) + return last_status + + +def _cancel_order_until_terminal(broker, order, *, timeout=30, retry_interval=2): + identifier = getattr(order, "identifier", None) + if not identifier: + raise AssertionError("Cannot clean up a broker order without an identifier") + + deadline = time.time() + timeout + last_status = None + last_error = None + while time.time() < deadline: + try: + last_status = _pull_order_status(broker, identifier) + last_error = None + except Exception as exc: + last_error = exc + + if last_status in _TERMINAL_STATUSES: + return last_status + + try: + broker.cancel_order(order) + except Exception as exc: + last_error = exc + + time.sleep(min(retry_interval, max(0, deadline - time.time()))) + + try: + last_status = _pull_order_status(broker, identifier) + last_error = None + except Exception as exc: + last_error = exc + if last_status in _TERMINAL_STATUSES: + return last_status + + message = ( + f"Broker order {identifier} did not reach a terminal status within {timeout} seconds; " + f"last status: {last_status!r}" + ) + if last_error is not None: + raise AssertionError(message) from last_error + raise AssertionError(message) + + +def _alpaca() -> Alpaca: + if not ALPACA_TEST_CONFIG.get("API_KEY") or not ALPACA_TEST_CONFIG.get("API_SECRET"): + pytest.skip("Missing ALPACA_TEST_API_KEY / ALPACA_TEST_API_SECRET in .env") + + return Alpaca( + dict(ALPACA_TEST_CONFIG), + max_workers=1, + connect_stream=False, + start_orders_thread=False, + ) + + +def _tradier() -> Tradier: + if not TRADIER_TEST_CONFIG.get("ACCOUNT_NUMBER") or not TRADIER_TEST_CONFIG.get("ACCESS_TOKEN"): + pytest.skip("Missing TRADIER_TEST_ACCOUNT_NUMBER / TRADIER_TEST_ACCESS_TOKEN in .env") + + return Tradier( + config=dict(TRADIER_TEST_CONFIG), + max_workers=1, + connect_stream=False, + ) + + +def _stock_limit_order(strategy_name): + return Order( + strategy_name, + Asset("AAPL", asset_type=Asset.AssetType.STOCK), + quantity=1, + side=Order.OrderSide.BUY, + limit_price=0.01, + time_in_force="gtc", + order_type=Order.OrderType.LIMIT, + ) + + +def _assert_direct_broker_order_lifecycle(broker, *, strategy_name): + submitted = None + try: + submitted = broker._submit_order(_stock_limit_order(strategy_name)) + assert submitted is not None + assert submitted.identifier, "Broker _submit_order() did not return an order id" + + status_before_cancel = _wait_for_broker_status(broker, submitted.identifier, timeout=30) + assert status_before_cancel in _LIVE_ACTIVE_STATUSES, ( + f"Unexpected broker status before cancel: {status_before_cancel!r}" + ) + + status_after_cancel = _cancel_order_until_terminal(broker, submitted) + assert status_after_cancel in _CANCELLED_STATUSES, ( + f"Broker order was not canceled: {status_after_cancel!r}" + ) + finally: + if submitted is not None and submitted.identifier: + _cancel_order_until_terminal(broker, submitted) + + +class _GtcLimitSubmitCancelStrategy(Strategy): + def initialize(self, parameters=None): + self.sleeptime = "1S" + self.submitted_order = None + self.submitted_identifier = None + self.status_before_cancel = None + self.status_after_cancel = None + self.cancel_requested = False + self.cancel_error = None + self.cash_seen = False + self.positions_seen = False + self.iteration_ran = False + self.strategy_end_ran = False + + def before_starting_trading(self): + self.cash_seen = self.get_cash() is not None + self.positions_seen = self.get_positions() is not None + + def on_trading_iteration(self): + asset = Asset(self.parameters["symbol"], asset_type=Asset.AssetType.STOCK) + order = self.create_order( + asset, + self.parameters["quantity"], + Order.OrderSide.BUY, + limit_price=self.parameters["limit_price"], + order_type=Order.OrderType.LIMIT, + time_in_force="gtc", + ) + + self.submitted_order = self.submit_order(order) + self.submitted_identifier = getattr(self.submitted_order, "identifier", None) + if self.submitted_identifier: + self.status_before_cancel = _wait_for_broker_status( + self.broker, + self.submitted_identifier, + timeout=30, + ) + self.iteration_ran = True + + def on_strategy_end(self): + self.strategy_end_ran = True + self._cancel_submitted_order() + + def on_bot_crash(self, error): + self._cancel_submitted_order() + + def _cancel_submitted_order(self): + if self.submitted_order is None or not self.submitted_identifier: + return + self.cancel_requested = True + try: + self.status_after_cancel = _cancel_order_until_terminal( + self.broker, + self.submitted_order, + timeout=30, + ) + self.cancel_error = None + except Exception as exc: + self.cancel_error = repr(exc) + + +def _run_live_strategy(broker, *, name): + broker.is_market_open = lambda: True + strategy = _GtcLimitSubmitCancelStrategy( + broker=broker, + name=name, + benchmark_asset=None, + analyze_backtest=False, + should_backup_variables_to_database=False, + should_send_summary_to_discord=False, + parameters={ + "symbol": "AAPL", + "quantity": 1, + "limit_price": 0.01, + }, + ) + # This CI test validates the order lifecycle inside a real strategy run. It bypasses only the scheduler's + # market-hours gate so pull-request CI is not tied to US equity session timing. Alpaca run_once has an early + # UTC-hours precheck, so patch the executor gate directly instead of relying on broker.is_market_open. + strategy._executor._initialize_live_market_calendars_for_run_once = ( + lambda: setattr(strategy._executor, "_run_once_market_open_override", True) + ) + trader = Trader(logfile="", backtest=False) + trader.add_strategy(strategy) + try: + result = trader.run_all(run_once=True) + finally: + _ensure_strategy_order_terminal(strategy, broker) + return strategy, result + + +def _ensure_strategy_order_terminal(strategy, broker): + if strategy.submitted_order is None or not strategy.submitted_identifier: + return + if _normalized_order_status(strategy.status_after_cancel) in _TERMINAL_STATUSES: + return + + strategy.cancel_requested = True + try: + strategy.status_after_cancel = _cancel_order_until_terminal( + broker, + strategy.submitted_order, + ) + except Exception as exc: + if strategy.cancel_error is None: + strategy.cancel_error = repr(exc) + + +def _assert_strategy_order_lifecycle(strategy, result): + assert result is not None + assert strategy.cash_seen + assert strategy.positions_seen + assert strategy.iteration_ran + assert strategy.strategy_end_ran + assert strategy.submitted_order is not None + assert strategy.submitted_identifier, "Strategy submit_order() did not return a broker order id" + assert strategy.status_before_cancel in _LIVE_ACTIVE_STATUSES, ( + f"Unexpected broker status before cancel: {strategy.status_before_cancel!r}" + ) + assert strategy.cancel_requested + assert strategy.cancel_error is None + assert strategy.status_after_cancel in _CANCELLED_STATUSES, ( + f"Broker order was not canceled: {strategy.status_after_cancel!r}" + ) + + +def test_alpaca_paper_broker_submits_polls_and_cancels_gtc_limit_order(): + broker = _alpaca() + try: + _assert_direct_broker_order_lifecycle(broker, strategy_name="alpaca-paper-broker-live-ci") + finally: + broker.cleanup_streams() + + +def test_tradier_paper_broker_submits_polls_and_cancels_gtc_limit_order(): + broker = _tradier() + try: + _assert_direct_broker_order_lifecycle(broker, strategy_name="tradier-paper-broker-live-ci") + finally: + broker.cleanup_streams() + + +def test_alpaca_paper_strategy_run_submits_polls_and_cancels_gtc_limit_order(): + broker = _alpaca() + try: + strategy, result = _run_live_strategy(broker, name="alpaca-paper-strategy-live-ci") + _assert_strategy_order_lifecycle(strategy, result) + finally: + broker.cleanup_streams() + + +def test_tradier_paper_strategy_run_submits_polls_and_cancels_gtc_limit_order(): + broker = _tradier() + try: + strategy, result = _run_live_strategy(broker, name="tradier-paper-strategy-live-ci") + _assert_strategy_order_lifecycle(strategy, result) + finally: + broker.cleanup_streams() diff --git a/tests/test_broker_live_strategy_run_safety.py b/tests/test_broker_live_strategy_run_safety.py new file mode 100644 index 000000000..55f92830a --- /dev/null +++ b/tests/test_broker_live_strategy_run_safety.py @@ -0,0 +1,60 @@ +from types import SimpleNamespace + +from tests.test_broker_live_strategy_run_apitest import ( + _cancel_order_until_terminal, + _ensure_strategy_order_terminal, +) + + +class _BrokerWithDelayedCancellation: + def __init__(self): + self.cancel_calls = 0 + self.statuses = iter(["open", "pending_cancel", "canceled"]) + + def _pull_broker_order(self, identifier): + return {"id": identifier, "status": next(self.statuses, "canceled")} + + def cancel_order(self, order): + self.cancel_calls += 1 + + +def test_cancel_order_until_terminal_retries_pending_cancellation(): + broker = _BrokerWithDelayedCancellation() + order = SimpleNamespace(identifier="paper-order-1") + + status = _cancel_order_until_terminal(broker, order, timeout=1, retry_interval=0) + + assert status == "canceled" + assert broker.cancel_calls == 2 + + +def test_strategy_safety_cancel_skips_terminal_order(): + broker = _BrokerWithDelayedCancellation() + strategy = SimpleNamespace( + submitted_order=SimpleNamespace(identifier="paper-order-1"), + submitted_identifier="paper-order-1", + status_after_cancel="canceled", + cancel_requested=True, + cancel_error=None, + ) + + _ensure_strategy_order_terminal(strategy, broker) + + assert broker.cancel_calls == 0 + + +def test_strategy_safety_cancel_preserves_prior_error(): + broker = _BrokerWithDelayedCancellation() + broker.statuses = iter(["canceled"]) + strategy = SimpleNamespace( + submitted_order=SimpleNamespace(identifier="paper-order-1"), + submitted_identifier="paper-order-1", + status_after_cancel=None, + cancel_requested=True, + cancel_error="initial cancellation failed", + ) + + _ensure_strategy_order_terminal(strategy, broker) + + assert strategy.status_after_cancel == "canceled" + assert strategy.cancel_error == "initial cancellation failed" diff --git a/tests/test_data_entity.py b/tests/test_data_entity.py index 3ca89e721..8236baf2c 100644 --- a/tests/test_data_entity.py +++ b/tests/test_data_entity.py @@ -326,6 +326,322 @@ def test_strict_intraday_rejects_multi_minute_history_past_bucket_tolerance(self with pytest.raises(ValueError, match="after the available data's end"): data.get_bars(base_dt + timedelta(minutes=35), length=3, timestep="5m") + def test_strict_intraday_rejects_stale_native_minute_fast_path(self): + asset = Asset("BTC", asset_type=Asset.AssetType.CRYPTO) + tz = pytz.timezone("America/New_York") + base_dt = tz.localize(datetime(2026, 6, 23, 23, 0)) + dates = [base_dt + timedelta(minutes=i) for i in range(3)] + df = pd.DataFrame( + { + "open": [100.0, 101.0, 102.0], + "high": [101.0, 102.0, 103.0], + "low": [99.0, 100.0, 101.0], + "close": [100.5, 101.5, 102.5], + "volume": [1.0, 1.0, 1.0], + }, + index=dates, + ) + data = Data(asset, df, timestep="minute", quote=Asset("USD", asset_type=Asset.AssetType.FOREX)) + data.strict_end_check = True + + bars = data.get_bars(base_dt + timedelta(minutes=3), length=2, timestep="minute") + assert bars is not None + + with pytest.raises(ValueError, match="after the available data's end"): + data.get_bars(base_dt + timedelta(minutes=4), length=2, timestep="minute") + + def test_strict_intraday_rejects_stale_native_multi_minute_fast_path(self): + asset = Asset("BTC", asset_type=Asset.AssetType.CRYPTO) + tz = pytz.timezone("America/New_York") + base_dt = tz.localize(datetime(2026, 6, 23, 23, 0)) + dates = [base_dt + timedelta(minutes=5 * i) for i in range(3)] + df = pd.DataFrame( + { + "open": [100.0, 101.0, 102.0], + "high": [101.0, 102.0, 103.0], + "low": [99.0, 100.0, 101.0], + "close": [100.5, 101.5, 102.5], + "volume": [1.0, 1.0, 1.0], + }, + index=dates, + ) + data = Data(asset, df, timestep="minute", quote=Asset("USD", asset_type=Asset.AssetType.FOREX)) + data._native_timestep_quantity = 5 + data._native_timestep_unit = "minute" + data.strict_end_check = True + + with pytest.raises(ValueError, match="after the available data's end"): + data.get_bars(base_dt + timedelta(minutes=30), length=2, timestep="5m") + + def test_strict_native_option_preserves_sparse_history_but_rejects_after_end(self): + asset = Asset("SPX", asset_type=Asset.AssetType.OPTION) + tz = pytz.timezone("America/New_York") + base_dt = tz.localize(datetime(2026, 6, 23, 9, 30)) + dates = [base_dt, base_dt + timedelta(minutes=5), base_dt + timedelta(minutes=30)] + df = pd.DataFrame( + { + "open": [10.0, 10.5, 11.0], + "high": [10.5, 11.0, 11.5], + "low": [9.5, 10.0, 10.5], + "close": [10.25, 10.75, 11.25], + "volume": [1.0, 1.0, 1.0], + }, + index=dates, + ) + data = Data(asset, df, timestep="minute") + data.strict_end_check = True + + request_dt = base_dt + timedelta(minutes=10) + bars = data.get_bars(request_dt, length=1, timestep="minute") + + assert bars is not None + assert bars.iloc[-1]["close"] == 10.25 + assert bars.index.max() < request_dt + + with pytest.raises(ValueError, match="resolved to stale .*data refresh required"): + data.get_last_price(request_dt) + + with pytest.raises(ValueError, match="after the available data's end"): + data.get_bars(base_dt + timedelta(minutes=40), length=1, timestep="minute") + + def test_strict_native_index_uses_default_inside_frame_tolerance(self): + asset = Asset("SPX", asset_type=Asset.AssetType.INDEX) + base_dt = pytz.timezone("America/New_York").localize(datetime(2026, 6, 23, 9, 30)) + dates = [base_dt, base_dt + timedelta(minutes=1), base_dt + timedelta(minutes=10)] + df = pd.DataFrame( + { + "open": [100.0, 101.0, 110.0], + "high": [101.0, 102.0, 111.0], + "low": [99.0, 100.0, 109.0], + "close": [100.5, 101.5, 110.5], + "volume": [1.0, 1.0, 1.0], + }, + index=dates, + ) + data = Data(asset, df, timestep="minute") + data.strict_end_check = True + + bars = data.get_bars(base_dt + timedelta(minutes=4), length=1, timestep="minute") + assert bars is not None + + with pytest.raises(ValueError, match="resolved to stale .*data refresh required"): + data.get_bars(base_dt + timedelta(minutes=5), length=1, timestep="minute") + + def test_strict_native_crypto_uses_default_inside_frame_tolerance(self): + asset = Asset("BTC", asset_type=Asset.AssetType.CRYPTO) + base_dt = pytz.timezone("America/New_York").localize(datetime(2026, 6, 23, 23, 0)) + dates = [base_dt, base_dt + timedelta(minutes=1), base_dt + timedelta(minutes=30)] + df = pd.DataFrame( + { + "open": [100.0, 101.0, 110.0], + "high": [101.0, 102.0, 111.0], + "low": [99.0, 100.0, 109.0], + "close": [100.5, 101.5, 110.5], + "volume": [1.0, 1.0, 1.0], + }, + index=dates, + ) + data = Data(asset, df, timestep="minute", quote=Asset("USD", asset_type=Asset.AssetType.FOREX)) + data.strict_end_check = True + + bars = data.get_bars(base_dt + timedelta(minutes=16), length=1, timestep="minute") + assert bars is not None + + with pytest.raises(ValueError, match="resolved to stale .*data refresh required"): + data.get_bars(base_dt + timedelta(minutes=17), length=1, timestep="minute") + + def test_strict_native_crypto_multi_minute_applies_sparse_gap_tolerance(self): + asset = Asset("BTC", asset_type=Asset.AssetType.CRYPTO) + tz = pytz.timezone("America/New_York") + base_dt = tz.localize(datetime(2026, 6, 23, 23, 0)) + dates = [ + base_dt, + base_dt + timedelta(minutes=5), + base_dt + timedelta(minutes=10), + base_dt + timedelta(minutes=30), + ] + df = pd.DataFrame( + { + "open": [100.0, 101.0, 102.0, 103.0], + "high": [101.0, 102.0, 103.0, 104.0], + "low": [99.0, 100.0, 101.0, 102.0], + "close": [100.5, 101.5, 102.5, 103.5], + "volume": [1.0, 1.0, 1.0, 1.0], + }, + index=dates, + ) + data = Data(asset, df, timestep="minute", quote=Asset("USD", asset_type=Asset.AssetType.FOREX)) + data._native_timestep_quantity = 5 + data._native_timestep_unit = "minute" + data.strict_end_check = True + + bars = data.get_bars(base_dt + timedelta(minutes=25), length=2, timestep="5m") + assert bars is not None + + with pytest.raises(ValueError, match="resolved to stale .*data refresh required"): + data.get_bars(base_dt + timedelta(minutes=26), length=2, timestep="5m") + + def test_native_minute_rejects_request_before_start_without_row_lookup(self, monkeypatch): + asset = Asset("SPY") + tz = pytz.timezone("America/New_York") + base_dt = tz.localize(datetime(2026, 6, 23, 9, 30)) + dates = [base_dt + timedelta(minutes=i) for i in range(3)] + df = pd.DataFrame( + { + "open": [100.0, 101.0, 102.0], + "high": [101.0, 102.0, 103.0], + "low": [99.0, 100.0, 101.0], + "close": [100.5, 101.5, 102.5], + "volume": [1.0, 1.0, 1.0], + }, + index=dates, + ) + data = Data(asset, df, timestep="minute") + + def forbidden_lookup(*args, **kwargs): + pytest.fail("invalid native request mutated the row cursor") + + monkeypatch.setattr(data, "get_iter_count", forbidden_lookup) + + with pytest.raises(ValueError, match="outside of the data's date range"): + data.get_bars(base_dt - timedelta(minutes=1), length=1, timestep="minute") + + def test_native_minute_rejects_invalid_length_and_timeshift_before_row_lookup(self, monkeypatch): + asset = Asset("SPY") + tz = pytz.timezone("America/New_York") + base_dt = tz.localize(datetime(2026, 6, 23, 9, 30)) + dates = [base_dt + timedelta(minutes=i) for i in range(2)] + df = pd.DataFrame( + { + "open": [100.0, 101.0], + "high": [101.0, 102.0], + "low": [99.0, 100.0], + "close": [100.5, 101.5], + "volume": [1.0, 1.0], + }, + index=dates, + ) + data = Data(asset, df, timestep="minute") + + def forbidden_lookup(*args, **kwargs): + pytest.fail("invalid native request performed a row lookup") + + monkeypatch.setattr(data, "get_iter_count", forbidden_lookup) + + with pytest.raises(TypeError, match="Length must be an integer"): + data.get_bars(base_dt, length="1", timestep="minute") + with pytest.raises(TypeError, match="Timeshift must be a number"): + data.get_bars(base_dt, length=1, timestep="minute", timeshift="1") + + def test_native_minute_naive_request_matches_timezone_validation_error(self, monkeypatch): + asset = Asset("SPY") + tz = pytz.timezone("America/New_York") + base_dt = tz.localize(datetime(2026, 6, 23, 9, 30)) + dates = [base_dt + timedelta(minutes=i) for i in range(2)] + df = pd.DataFrame( + { + "open": [100.0, 101.0], + "high": [101.0, 102.0], + "low": [99.0, 100.0], + "close": [100.5, 101.5], + "volume": [1.0, 1.0], + }, + index=dates, + ) + data = Data(asset, df, timestep="minute") + + def forbidden_lookup(*args, **kwargs): + pytest.fail("timezone-invalid request performed a row lookup") + + monkeypatch.setattr(data, "get_iter_count", forbidden_lookup) + + with pytest.raises(TypeError, match="offset-naive and offset-aware"): + data.get_bars(datetime(2026, 6, 23, 9, 31), length=1, timestep="minute") + + def test_native_daily_bar_covers_utc_calendar_date(self): + dates = pd.to_datetime(["2026-11-02 00:00:00", "2026-11-03 00:00:00"], utc=True) + df = pd.DataFrame( + { + "open": [100.0, 101.0], + "high": [101.0, 102.0], + "low": [99.0, 100.0], + "close": [100.5, 101.5], + "volume": [1.0, 1.0], + }, + index=dates, + ) + data = Data(Asset("SPY"), df, timestep="day") + request_dt = pytz.timezone("America/New_York").localize(datetime(2026, 11, 3, 8, 30)) + + bars = data.get_bars(request_dt, length=1, timestep="day") + + assert bars is not None + assert bars.iloc[-1]["close"] == 101.5 + + def test_native_minute_normalizes_timedelta_timeshift(self): + asset = Asset("SPY") + tz = pytz.timezone("America/New_York") + base_dt = tz.localize(datetime(2026, 6, 23, 9, 30)) + dates = [base_dt + timedelta(minutes=i) for i in range(4)] + df = pd.DataFrame( + { + "open": [100.0, 101.0, 102.0, 103.0], + "high": [101.0, 102.0, 103.0, 104.0], + "low": [99.0, 100.0, 101.0, 102.0], + "close": [100.5, 101.5, 102.5, 103.5], + "volume": [1.0, 1.0, 1.0, 1.0], + }, + index=dates, + ) + data = Data(asset, df, timestep="minute") + + bars = data.get_bars( + base_dt + timedelta(minutes=3), + length=2, + timestep="minute", + timeshift=timedelta(minutes=1), + ) + + assert bars is not None + assert list(bars["close"]) == [100.5, 101.5] + + def test_native_minute_hot_path_reuses_iter_count_and_integer_gap(self, monkeypatch): + asset = Asset("SPY") + tz = pytz.timezone("America/New_York") + base_dt = tz.localize(datetime(2026, 6, 23, 9, 30)) + dates = [base_dt + timedelta(minutes=i) for i in range(3)] + df = pd.DataFrame( + { + "open": [100.0, 101.0, 102.0], + "high": [101.0, 102.0, 103.0], + "low": [99.0, 100.0, 101.0], + "close": [100.5, 101.5, 102.5], + "volume": [1.0, 1.0, 1.0], + }, + index=dates, + ) + data = Data(asset, df, timestep="minute") + data.strict_end_check = True + original_get_iter_count = data.get_iter_count + iter_count_calls = 0 + + def tracked_get_iter_count(dt): + nonlocal iter_count_calls + iter_count_calls += 1 + return original_get_iter_count(dt) + + def forbidden_fallback(*args, **kwargs): + pytest.fail("native hot path used generic or pandas-based validation") + + monkeypatch.setattr(data, "get_iter_count", tracked_get_iter_count) + monkeypatch.setattr(data, "_validate_bars_request", forbidden_fallback) + monkeypatch.setattr(data, "_strict_intraday_stale_bar_error", forbidden_fallback) + + bars = data.get_bars(base_dt + timedelta(minutes=2), length=2, timestep="minute") + + assert bars is not None + assert iter_count_calls == 1 + def test_large_tz_aware_repair_avoids_retained_iter_index_dict_and_preserves_lookup(self): asset = Asset("MEM") index = pd.date_range("2024-01-01", periods=50_001, freq="min", tz="America/New_York") diff --git a/tests/test_ibkr_futures_daily_series.py b/tests/test_ibkr_futures_daily_series.py index 4e5f374d1..b2068223a 100644 --- a/tests/test_ibkr_futures_daily_series.py +++ b/tests/test_ibkr_futures_daily_series.py @@ -3,13 +3,13 @@ from datetime import date, datetime, timezone import pandas as pd -import pytest from lumibot.entities import Asset def test_ibkr_futures_daily_bars_are_session_aligned_not_midnight(monkeypatch): import pandas_market_calendars as mcal + import lumibot.tools.ibkr_helper as ibkr_helper fut = Asset("MES", asset_type=Asset.AssetType.FUTURE, expiration=date(2025, 12, 19)) @@ -37,7 +37,17 @@ def test_ibkr_futures_daily_bars_are_session_aligned_not_midnight(monkeypatch): calls: list[str] = [] - def fake_get_cached_bars_for_source(*, asset, quote, timestep, start_dt, end_dt, exchange, include_after_hours, source): + def fake_get_cached_bars_for_source( + *, + asset, + quote, + timestep, + start_dt, + end_dt, + exchange, + include_after_hours, + source, + ): calls.append(str(timestep)) if str(timestep) == "hour": return intraday diff --git a/tests/test_tradovate.py b/tests/test_tradovate.py index 1ead58ca7..ae0275851 100644 --- a/tests/test_tradovate.py +++ b/tests/test_tradovate.py @@ -277,8 +277,8 @@ def test_broker_connect_on_init_preserves_eager_connection(self): assert broker.account_id == 123456 assert broker.user_id == 'fake_user_id' - def test_lazy_polling_waits_for_explicit_connection(self): - """Polling stream must not authenticate a lazy Tradovate broker in the background.""" + def test_lazy_connection_starts_requested_polling_stream_once(self): + """The default live path starts polling after authentication, not during construction.""" from lumibot.brokers import Tradovate config = { @@ -289,17 +289,54 @@ def test_lazy_polling_waits_for_explicit_connection(self): "IS_PAPER": True, } - with patch.object(Tradovate, '_get_tokens') as mock_get_tokens, \ - patch.object(Tradovate, '_get_account_info') as mock_get_account_info, \ - patch.object(Tradovate, '_get_user_info') as mock_get_user_info: - broker = Tradovate(config=config, connect_stream=True) - assert hasattr(broker, "stream") + tokens = { + "accessToken": "token", + "marketToken": "market", + "hasMarketData": True, + } + stream = MagicMock() + with patch.object(Tradovate, '_get_tokens', return_value=tokens), \ + patch.object(Tradovate, '_get_account_info', return_value={"accountSpec": "TEST", "accountId": 123}), \ + patch.object(Tradovate, '_get_user_info', return_value="user"), \ + patch.object(Tradovate, '_get_stream_object', return_value=stream) as mock_get_stream, \ + patch.object(Tradovate, '_launch_stream') as mock_launch_stream: + broker = Tradovate(config=config) + assert not hasattr(broker, "stream") - broker.do_polling() + broker._ensure_connected() + broker._ensure_connected() - mock_get_tokens.assert_not_called() - mock_get_account_info.assert_not_called() - mock_get_user_info.assert_not_called() + assert broker.stream is stream + mock_get_stream.assert_called_once_with() + mock_launch_stream.assert_called_once_with() + + def test_connect_stream_false_keeps_polling_disabled_after_connection(self): + """Scheduled/run-once callers can explicitly keep polling disabled.""" + from lumibot.brokers import Tradovate + + config = { + "USERNAME": "test_user", + "DEDICATED_PASSWORD": "test_pass", + "CID": "test_cid", + "SECRET": "test_secret", + "IS_PAPER": True, + } + tokens = { + "accessToken": "token", + "marketToken": "market", + "hasMarketData": True, + } + with patch.object(Tradovate, '_get_tokens', return_value=tokens), \ + patch.object(Tradovate, '_get_account_info', return_value={"accountSpec": "TEST", "accountId": 123}), \ + patch.object(Tradovate, '_get_user_info', return_value="user"), \ + patch.object(Tradovate, '_get_stream_object') as mock_get_stream, \ + patch.object(Tradovate, '_launch_stream') as mock_launch_stream: + broker = Tradovate(config=config, connect_stream=False) + broker._ensure_connected() + + assert not hasattr(broker, "stream") + mock_get_stream.assert_not_called() + mock_launch_stream.assert_not_called() def test_first_submit_order_connects_once(self): """Default lazy startup connects to Tradovate only when an authenticated operation runs."""