Files
OpenBB/openbb_platform/providers/sec/tests/test_sec_financials.py
Danglewood bbf1ab2020 V5 (#7489)
* [Feature] OpenBB V5: Initial `openbb-core` Refactor (#7482)

* initial refactor

* ruff does not dictate style

* unresolved imports in test action

* fix general linting

* resolve pandas imports for actions run

* httpx in test workflow

* pytest-asyncio in test workflow

* pandas.Index(...)

* max version for windows CI is 3.14

* remove temp refactor files that shouldn't have been committed:

* use ruff format instead of black

* formatting and typing

* tab space helper

* test fixtures..

* ty check

* also run integration test in CI

* remove that annoying labeler action

* [V5] `openbb-core`: 100% Test Coverage (#7484)

* increase test coverage to exercise 100% of the testable code path in openbb-core

* model_fields syntax cleanup

* set the environment variable for CI runs

* run codeql on v5 branch

* more CI env setup

* cleanup redundant files

* ruff

* dedupe workflow trigger

* some windows coverage missing

* ubuntu integration test job

* more windows coverage

* test_reference_generator

* test_docstring_generator

* remove line no references from test function docstrings

* clean useless comments from test files

* fix polars type check

* fix test_obbject

* fix test_obbject

* package builder fixes

* charting pyproject version

* fix docstring generator for nested models

* missing coverage

* [V5] `openbb-core`: Add Backwards-Compatible Layered Config System (openbb.toml) (#7492)

* add backwards-compatible layered config system

* fix path reference in windows test

* need python version-aware handling in docstring generator

* update for deprecated pydantic syntax

* ruff UP007

* test_system_settings

* remove v4 platform workflow

* [V5] Refactor `openbb-oecd` for full SDMX 2 API Exposure and V5 Packaging (#7413)

* stash some changes

* add more robust testing

* mypy

* point PR at V5

* introduce spec file

* codespell

* test fix

* fix workflow environment setup

* fix workflow environment setup

* fix workflow environment setup

* add pyyaml to dependencies

* split lint jobs

* fix workflow environment setup

* fix workflow environment setup

* workflow env setup

* workflow env setup

* clean up code comments

* add auth hook entrypoints

* codespell

* add codegen feature

* codespell

* move _unpack into dispatchers for consistency with codegen packages

* surface nested models in the response

* fix missing coverage in CI

* socrata updates

* test fix

* detect plotly output

* add --include and --exclude flags from generate-extension command

* refactor openbb-platform-api for V5 operations and spec file handling

* delete poetry.lock

* add test workflow

* add temp branch id for test workflow

* add httpx to dev dependencies

* add a conftest.py

* tweak conftest

* tweak conftest

* python 3.10 issues

* more python 3.10

* more python 3.10

* more python 3.10

* more python 3.10

* fix url template resolution

* multi-spec support

* fix a windows path issue

* merge branch develop

* table builder, integration tests, and more

* linting

* merge branch develop

* bit of cleanup

* some more updates, cleanup, and apps.json

* codespell

* black

* integration test params and linting

* mypy

* pylint

* black

* more pylint

* recapture teest cassettes

* black again

* more linting

* table presentation choices

* black

* fix merge conflict

* split up metadata module and reformat cache to use msgpack

* mypy

* regenerate lock file

* use json and include dependencies to make openbb-oecd a standalone install

* pylint

* more linting

* serve metadata as dependency instead of assign and initialize

* fix integration_tests_testers to not count dependency injections as paramters

* add assets to git ignore

* ruff

* v5 refactor and publish workflow

* test coverage and ty fixes

* codespell

* update lock file

* workflow setup

* ty check

* workflow setup

* Scope PR to OECD changes only

Restore non-OECD files in the diff against v5 to their v5 state. The
branch was originally created off feature/cli-2 and inherited 186 files
of unrelated changes (cli, agent_server, mcp_server, platform_api,
core/, .github/workflows/* not oecd). Those have nothing to do with the
openbb-oecd v5 refactor and were polluting the PR diff.

* scope cleanup: include leaked v5-resident files (general-linting workflow, platform_api test fixtures)

---------

Co-authored-by: deeleeramone <>
Co-authored-by: Copilot <copilot@github.com>

* fix docstring generator and body param unwrapping

* project ty.toml file

* unpack POST parameters from params: Model

* remove comment chatter

* [V5] `openbb-oecd`: Rename `oecd_utils` as `oecd` (#7509)

* rename oecd_utils to oecd and populate all models there when openbb-economy is not installed

* linting

* bump dependencies

* fix test drift

* linter

* FieldInfo import

* [V5] `openbb-imf`: Refactor for V5 (#7508)

* refactor imf extension for V5

* codespell

* fix some python 3.10 items

* return self on copy of metadata object

* [V5] `openbb-news`: Refactor News for V5 and Add Configurable RSS Endpoint (#7507)

* refactor news for v5 and add rss endpoint

* remove poetry.lock

---------

Co-authored-by: deeleeramone <>

* [V5] `openbb-quantitative`: Refactor for V5 (#7506)

* refactor quantitative for v5

* add factors endpoint and add a few charts

---------

Co-authored-by: deeleeramone <>

* [V5] `openbb-technical`:  Refactor for V5 & Add More Quick Win Endpoints (#7502)

* refactor openbb-technical for v5

* forward refs

* linting

* no python 3.15 in test matrix

* reame

* ty

* move all params to body

---------

Co-authored-by: deeleeramone <>

* remove stockgrid provider (#7486)

* [V5] remove wsj provider and related code (#7485)

* remove wsj provider and related code

* purge from metadata

* actions

* actions

* openbb-core publish workflow

* delete biztoc (#7523)

* [V5] Remove `openbb-alpha-vantage` (#7522)

* delete alpha vantage

* more files

---------

Co-authored-by: Theodore Aptekarev <aptekarev@gmail.com>

* [V5] Remove `openbb-benzinga` (#7521)

* delete benzinga

* more references

---------

Co-authored-by: Theodore Aptekarev <aptekarev@gmail.com>

* [V5] `openbb-bls`: Refactor for V5 + Add Many Endpoints (#7518)

* refactor bls for v5, add 100 widgets

* some CI fixes

* no python 3.15

* fix missing dev group in pyproject.toml

* more pyproject fallout

* action

* refactor econometrics for v5 (#7505)

Co-authored-by: deeleeramone <>

* [V5] `openbb-famafrench`: Refactor for V5 (#7503)

* refactor famafrench for v5

* codespell

* finishing touches

---------

Co-authored-by: deeleeramone <>

* [V5] CLI 2.0 - Pluggable Backends, Non-TTY Default, HTTP Dispatcher, Spec Files + Codegen (#7433)

* stash some changes

* add more robust testing

* mypy

* point PR at V5

* introduce spec file

* codespell

* test fix

* fix workflow environment setup

* fix workflow environment setup

* fix workflow environment setup

* add pyyaml to dependencies

* split lint jobs

* fix workflow environment setup

* fix workflow environment setup

* workflow env setup

* workflow env setup

* clean up code comments

* add auth hook entrypoints

* codespell

* add codegen feature

* codespell

* move _unpack into dispatchers for consistency with codegen packages

* surface nested models in the response

* fix missing coverage in CI

* socrata updates

* test fix

* detect plotly output

* add --include and --exclude flags from generate-extension command

* cap test matrix at python 3.14

* no useless comments

* platform controller command description split

* merge URL overloads from path params

* exclude none and unset from model dump

---------

Co-authored-by: deeleeramone <>
Co-authored-by: Copilot <copilot@github.com>

* general-linting

* downgrade aiohttp in devtools for vcrpy

* some ty items and add requests-types stubs for ty in devtools

* markdown linting

* remove integration tests from extensions that were moved into the famafrench provider

* [V5] `openbb-congress-gov`: Refactor for V5 + Convert To Using Bulk Files Instead of `api.congress.gov`. (#7528)

* refactor congress-gov for v5 and use bulk files instead of API with credentials

* integration tests

* apps tab name

* convert caching to sql

* codespell

* [V5] [BugFix] Fix for Breaking Changes in FastAPI 0.137+ (#7537)

* fix route introspection for fastapi 0.137

* recapture oeccd gdp forecast test cassette

---------

Co-authored-by: deeleeramone <>

* [V5] `openbb-platform-api`: Refactor for V5 Spec, Config, Packaging, Test Coverage (#7494)

* stash some changes

* add more robust testing

* mypy

* point PR at V5

* introduce spec file

* codespell

* test fix

* fix workflow environment setup

* fix workflow environment setup

* fix workflow environment setup

* add pyyaml to dependencies

* split lint jobs

* fix workflow environment setup

* fix workflow environment setup

* workflow env setup

* workflow env setup

* clean up code comments

* add auth hook entrypoints

* codespell

* add codegen feature

* codespell

* move _unpack into dispatchers for consistency with codegen packages

* surface nested models in the response

* fix missing coverage in CI

* socrata updates

* test fix

* detect plotly output

* add --include and --exclude flags from generate-extension command

* refactor openbb-platform-api for V5 operations and spec file handling

* delete poetry.lock

* add test workflow

* add temp branch id for test workflow

* add httpx to dev dependencies

* add a conftest.py

* tweak conftest

* tweak conftest

* python 3.10 issues

* more python 3.10

* more python 3.10

* more python 3.10

* more python 3.10

* fix url template resolution

* multi-spec support

* fix a windows path issue

* cap test matrix at python 3.14

* no 3.15 runners for platform-api

* no useless comments

* platform controller command description split

* merge URL overloads from path params

* exclude none and unset from model dump

* markdown lint

* ignore get/post endpoints that are return SSE

* fix remaining fastapi 0.137 fallout

---------

Co-authored-by: deeleeramone <>
Co-authored-by: Copilot <copilot@github.com>

* add full support for AsyncIterator response type for streaming output (#7530)

* [V5] Refactor Shared Pass Through Router Extensions For Hatchling (#7524)

* refactor shared pass-through routers for hatchling and ruff + ty

* add publish workflows

* [V5] `openbb-charting`: Refactor for V5 and Decouple  (#7545)

* refactor charting for v5 patterns and decouple entirely

* coveragerc

* lint and tests

* fix tests

* configure dev group with openbb-technical

* ty

* fix cli test

* remove 3.15 from test matrix - dependencies are unstable

---------

Co-authored-by: deeleeramone <>

* fix merge widgets

* workflow cleanup"

* fixes for ty

* fogot that file

* general lint setup

* consolidate lint job

* remove seeking alpha provider (#7570)

* remove tradier provider (#7573)

* remove tiingo provider (#7574)

* remove intrinio provider (#7575)

Remove the openbb-intrinio provider extension from the repository.
Reason: proprietary vendor with no internal access to any endpoint;
cannot build, develop or maintain.

* remove yfinance provider from repo (#7576)

* fix(cli): resolve modular OpenAPI references (#7586)

Fixes #7585

* [V5] `openbb-cli`: Modular OpenAPI Refs (#7589)

* follow up to #7585

* add another codegen test

* unused ty ignore

* Update generated tests to get python parameter names instead of raw API-provider parameter names. (#7597)

* Update the record-vs-envelope heuristic in the payload processing code to not treat a record that happens to have a list datatype as an entire envelope. (#7598)

* Update class generation code to handle empty data that php-style serialization emits as []. (#7599)

* fix ty check failures exposed by newer version

* add drugs.com and brutalist as news rss feeds (#7602)

Co-authored-by: deeleeramone <>

* codespell

* missing test coverage that appeared in famafrench..

* plotly 7 handling for mapbox namespace change

* remove finviz (#7577)

* remove trading economics provider (#7578)

* remove openbb-multpl from repository (#7610)

* add httpx2 to devtools for starlette test client

* restore test coverage to 100% in openbb_core

* remove fmp provider extension from repository (#7665)

* unpin ruff

* fix charting conftest and missing coverage in 3.10

* add charting coverage

* fix new ty issues

* fix tag grouping in swagger docs

* [V5] `openbb-ecb`: Refactor for V5 and Expand Coverage (#7560)

* refactor openbb-ecb for v5 and expand coverage

* some updates

* tests

* workflow

* remove 3.15 from test matrix

* some more table schema updates

* remove some html chrome on embedded articles

---------

Co-authored-by: deeleeramone <>

* [V5] `openbb-federal-reserve`: Refactor for V5 + FFIEC & Regional Fed Routers (#7561)

* refactor for v5

* add workflows

* fix bhcpr schema

* fix pyproject.toml

* fix test

* codespell

* ty

* fix pyproject

* fix ci environment

* fix ci environment

* convert documents endpoints to use fedinprint service

* codeql fix

* fix ny publications and add ny nowcast

* [V5] `openbb-eia`: Refactor for V5 + Add Full API Implementation (#7584)

* eia refactor for v5

* add test and publish workflows

* data browser state

* fix workflows

* more workflow test items

* temp files in tests

* more fixes

* more browsers

* fix codeql issues

* fix mer state and ensure multi-user capability

* uv lock

* table bridge

* fix electricity browser state

* convert HTML widgets to iFrame

* fix recursion error in workspace mcp

* [V5] `openbb-government-us`: Refactor for V5 + Move `uscongress` (#7604)

* v5 government_us provider

* workflow files

* ty and add xlrd to pyproject

* shrink test cassettes

* dev dependencies

* wheel packaging and more tests

* python 3.10 thing..?

* fix test

* default state for multiselect dropdowns

* replace congress html widgets with iframes

* fallback to congress api for congress before 108

* ensure member passage voting count populates

* [V5] Feature: Add JODI World Oil & Gas Database (#7593)

* add the JODI oil and gas database as a provider/router extension

* lint fixes

* add grouping to apps.json

---------

Co-authored-by: deeleeramone <>

* [V5] `openbb-cftc`: Refactor for V5 + Add OIS, Forward Curve, and FX Trade Analysis (#7615)

* refactor openbb-cftc for v5

* add SHIBOR to fixings

---------

Co-authored-by: deeleeramone <>

* fix for plotly 7 breaking change to schemas

* [V5] `openbb-tmx`: Refactor extension for V5 (#7668)

* refactor tmx provider for v5

* fix tests..

* fix tests stack trace exposure

* fix workflow setup

* [V5] `openbb-nasdaq`: Refactor provider for V5. (#7670)

* refactor nasdaq provider for v5

* add workflows

* ty fix

* [V5] `openbb-fred`: Refactor as V5 extension. (#7669)

* refactor fred extension for v5

* fix markdown MD037 lint error in README

* codespell

* [V5] `openbb-cboe`: Refactor for V5 + Add Payoff Diagrams (#7617)

* refactor cboe for v5

* use zoneinfo instead

* fix test

* some fixes related to the cache access

* fix a couple of tests

* fix the test coverage

* [V5] `openbb-deribit`: Refactor provider for V5 (#7674)

* refactor openbb-deribit for v5

* restructure the optimizer around the published combos data and add integration test suites

* fix markdown lint errors

* remove openbb-econdb provider (#7675)

* codespell

* [V5] `openbb-sec`: Refactor for V5 (#7525)

* refactor openbb-sec for v5

* linting

* ty

* normalize compare_company_facts

* add workflows

* python 3.10 fix

* add more thorough url validation and tests

* ty check

* ty for python 3.10

* reconfigure with apps

* temp pyproject overrride

* pyproject

* pyproject

* pyproject

* codespell

* pyproject

* fix uk schema file links in schema_files

* test coverage

* test_cli

* port #7547 into this pr

* no hard coded prefix

* improve filing rendering capabilities

* fix xml text detection

* make codeql happy

* missed a file in the commit

* test coverage

* codespell

* fix flakey test

* some updates

* management_ownership --> management_profiles

* proxy statement markdown

* ruff

* fix test coverage

* more tests

* test coverage

* fix lint errors

* Add SEC adviser search, profile, and universe routes (#7559)

* Add SEC institutional data fetchers

* Add SEC filing index and Form D routes

* Add SEC investment adviser people data

* Expose SEC adviser and institutional routes

* fix(sec): address adviser review feedback

* refactor(sec): narrow adviser support to IAPD search

* feat(sec): add adviser firm and individual searches

* feat(sec): add adviser profile drill-down

* fix: repair pre-commit configuration

* feat(sec): add adviser universe route

* feat(sec): expose complete Form ADV data

---------

Co-authored-by: Danglewood <85772166+deeleeramone@users.noreply.github.com>

* fix test coverage

---------

Co-authored-by: deeleeramone <>
Co-authored-by: Theodore Aptekarev <aptekarev@gmail.com>

* [V5] `openbb-finra`:  Refactor for V5 (#7676)

* refactor finra provider for v5

* add workflows

* fix tests

* fix lint errors

* [V5] Update Repository License to Apache 2.0 (#7677)

* change license to Apache 2.0 for V5

* codespell

* bump devtools and repoint pyproject.toml

* Enterprise Flask Migration Toolkit (#7264)

* feat: Add Flask-to-OpenBB enterprise migration toolkit

- Implement Router.from_flask() for zero-touch legacy API conversion
- Enable 4-hour migration vs 6-month traditional rewrite (80% cost reduction)
- Generate FastAPI endpoints from Flask routes with automatic type validation
- Support enterprise consulting services and custom integration workflows

Demonstrates OpenBB's competitive advantage for institutional client migrations.
Complements existing Workspace applications like Open Portfolio suite.

* chore: Resolve .gitignore conflict and add common ignores

* refactor: implement Phase 1 Flask integration architecture- Move Flask conversion logic to openbb_core.app.utils.flask- Add Flask detection to extension_loader - Integrate with OpenBB's core extension system- Add Flask dependency validation with sys.modules checkingAddresses @deeleeramone's architectural feedback for native OpenBB integration.

* docs: add Flask direct entry point demos- demo_direct_flask_entry_point.py: shows direct Flask instance reference- example_pyproject_flask_entry.toml: demonstrates pyproject.toml configurationSupports new architecture eliminating wrapper extensions.

* Implement Phase 1 Flask adapter per code review feedback

- Minimal working implementation with Flask detection and router creation
- Removes complex code generation logic for focused Phase 1 delivery
- Establishes foundation for Phase 2 route conversion features

* Fix Flask integration: Replace missing FlaskToOpenBBAdapter with WSGIMiddleware

- Remove FlaskToOpenBBAdapter which was not implemented
- Use FastAPI's WSGIMiddleware to mount Flask apps directly
- Update extension_loader to use returned Router object
- Convert utils.py to utils/__init__.py to support flask submodule
- Addresses PR review comment about unhandled AttributeError

* Address PR review feedback: fix Flask imports and remove unused files

- Revert .gitignore to upstream/develop
- Remove AI-generated demo/example files and flask_adapter extension
- Move Flask/werkzeug imports inside conditional blocks to avoid ImportError
- Use lazy imports via __getattr__ in utils/flask/__init__.py
- Add Flask tests with @pytest.mark.skipif markers

* Force APIRouter.include_router to propagate Mount paths instead of dropping them. (#1)

This change ensures that mounted ASGI applications (such as Flask apps wrapped in WSGIMiddleware) are not dropped when `APIRouter.include_router` is called.

Previously, while the initial load in `ExtensionLoader` worked as expected, the `Mount` objects were ignored by FastAPI's default inclusion logic. This caused the mounted elements to be lost during the final application assembly, resulting in 404 errors for the Flask routes.

This fix overrides the inclusion logic to manually propagate `Mount` objects downstream.

* feat(flask): Add metadata layer for OpenAPI documentation

Enhance Flask-to-OpenBB converter with structured docstring parsing:
- Parse Google-style docstrings (summary, description, params)
- Extract parameter descriptions from Args section
- Pass metadata to Router.command() for rich OpenAPI docs

This addresses PR feedback to provide tangible benefit by generating
comprehensive API documentation for migrated Flask routes.

* feat(flask): Implement manual OpenAPI spec generation and attachment

- Add FlaskParam and FlaskRouteInfo TypedDict for type-safe metadata
- Implement OpenAPISpecGenerator for Flask→OpenAPI conversion
- Extract route metadata via FlaskIntrospector
- Attach generated spec to router.openbb_openapi_spec_ext
- Add unit tests (3/3 passing)

Addresses WSGI/mount boundary issue where Router.command metadata
cannot propagate through WSGIMiddleware. Spec attached to router
instance for downstream OpenAPI assembly.

Resolves maintainer feedback on PR #7264

* feat(flask): Implement FlaskMountRegistry for multiple mount support

- Add FlaskMountRegistry class for centralized OpenAPI spec management
- Replace router.openbb_openapi_spec_ext with registry pattern
- Support multiple Flask apps at different mount prefixes
- Automatic path rewriting in aggregated OpenAPI specs
- Add registry test case (4/4 tests passing)

Addresses @deeleeramone's feedback on PR #7264 regarding router
attribute fragility and multiple mount support. Registry provides
lifecycle-independent state management and enables enterprise
scenarios with multiple Flask apps.

Resolves: OpenBB-finance/OpenBB#7264 (partial)

* feat(core): load Flask apps as openbb_core_extension WSGI sub-apps

A Flask app referenced directly by an openbb_core_extension entry point is
mounted via WSGIMiddleware and its routes are introspected into openapi.json
(operationId/widgetId/mcp_tool, parameters, request bodies, security, widget_config
and HTML widgets) so the widget factory and MCP server can discover them.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* test(core): 100% coverage for flask utils + flask dev dependency

Add branch-coverage tests for the Flask integration package (introspection,
AST analysis, docstrings, OpenAPI generation, registry, loader, detection) and
declare flask as a dev dependency so the suite runs in CI.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(flask): treat f-string returns as HTML responses

* fix(flask): mount via a2wsgi instead of deprecated starlette WSGIMiddleware

- add openbb-core[flask] extra (flask, a2wsgi); a2wsgi resolved through require_optional
- run the WSGI serve test through mount_flask_extensions; drop httpx importorskip
- core CI installs [pandas,flask] and httpx2

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test(flask): cover failed entry point loads; narrow optional types for ty

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Danglewood <85772166+deeleeramone@users.noreply.github.com>
Co-authored-by: deeleeramone <>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

* perf(core): vectorize the date conversion in `basemodel_to_df` (#7622)

* perf(core): vectorize the date conversion in basemodel_to_df

basemodel_to_df ran three per-element Python passes over the date column of
every payload it converted. Series.apply(to_datetime) calls pandas.to_datetime
once per row, and because each call receives a lone scalar with no format hint,
pandas also re-runs its datetime format guesser on every row.

Parse the column in one call instead, and use vectorized comparisons for the
all-midnight test and the reduction to datetime.date.

The element-wise parse is kept as a fallback for one case that the vectorized
path cannot represent: a column mixing UTC offsets, such as a series crossing a
daylight-savings boundary. Parsing per element lets each value keep its own
offset and yields an object column of tz-aware Timestamps, whereas a single
to_datetime call over that column raises "Mixed timezones detected". That
behaviour is covered by test_to_df_daylight_savings and is preserved exactly.

Measured on a payload built through EquityHistoricalData, median of 9 under an
exclusive CPU lock:

  250 daily bars     8.92 ms -> 1.46 ms
  1,000 daily bars  32.93 ms -> 3.13 ms
  5,000 daily bars 162.25 ms -> 12.23 ms

* fix a broken test - unrelated to this PR...

---------

Co-authored-by: Danglewood <85772166+deeleeramone@users.noreply.github.com>
Co-authored-by: deeleeramone <>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

* fix lint errors

* fix oecd lint error

* [V5] `openbb-mcp-server`: Refactor for v5 - Spec, Config, Packaging, Test Coverage (#7495)

* stash some changes

* add more robust testing

* mypy

* point PR at V5

* introduce spec file

* codespell

* test fix

* fix workflow environment setup

* fix workflow environment setup

* fix workflow environment setup

* add pyyaml to dependencies

* split lint jobs

* fix workflow environment setup

* fix workflow environment setup

* workflow env setup

* workflow env setup

* clean up code comments

* add auth hook entrypoints

* codespell

* add codegen feature

* codespell

* move _unpack into dispatchers for consistency with codegen packages

* surface nested models in the response

* fix missing coverage in CI

* socrata updates

* test fix

* detect plotly output

* add --include and --exclude flags from generate-extension command

* refactor mcp server for v5

* test action

* fix tests

* tests

* test fixture name

* revert: drop unrelated provider/README changes from mcp-server refactor

Strip the changes to ``README.md`` and ``openbb_platform/providers/{imf,
tradingeconomics,yfinance}`` that snuck into 4ac3602 (refactor mcp server
for v5). Those files are out of scope for the mcp-server work and were
modified inadvertently — restore them to their pre-4ac3602 state.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* tests and workflows

* fix workflow

* tests and workflows

* tests and workflows

* tests and workflows

* tests and workflows

* beartype shim for 3.15

* muck around with beartype for 3.15

* muck around with beartype for 3.15

* no 3.15 runners for mcp-server

* update skill files

* cap test matrix at python 3.14

* merge branch v5

* no useless comments

* platform controller command description split

* merge URL overloads from path params

* exclude none and unset from model dump

* refactor for fastmcp4

* wrap markdown lines that are too long

---------

Co-authored-by: deeleeramone <>
Co-authored-by: Copilot <copilot@github.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* update pre-commit, noxfile, and uv locks

* fixes for docstring generation

* test coverage

* fix cli renderer helper with figure object JSON

* fix the figure font when rendered

* refactor cookiecutter

* bump cookiecutter packagge version

* fix sec test by adding fixture

* fix desktop environments test

* fix(cli): inherit path-item-level OpenAPI parameters (#7606)

* fix(cli): inherit path-item-level OpenAPI parameters

OpenAPI 3.x lets a Path Item Object declare `parameters` shared by every
operation under that path, with operation-level entries overriding them by
(name, in). The codegen read only `op.get("parameters")`, so inherited
parameters were dropped silently — the generated command kept the URL
template but had nothing registered to fill it.

Real-world impact: the Xero Accounting API declares its required
`xero-tenant-id` header once per path item across all 138 paths. Before this
change `build_command_spec` produced 78 commands and none carried the
header, so every generated command 403s. Nothing warned.

Adds `operation_parameters()`, which merges the two levels and dereferences
both, and routes the pre-computed spec path (`build_command_spec`) and the
interactive path (`build_command_index` -> `build_parser_from_operation`)
through it. Unresolvable refs, non-dict entries, and nameless parameters are
skipped as before. Specs without path-item parameters are unaffected.

* fix(cli): narrow path-item inheritance to path and query parameters

Per review: headers do not translate into a Python interface function, and
outbound headers are transport configuration
(SystemSettings.PythonSettings.http.headers, the CLI's -H) or a security-scheme
concern — not a per-command argument.

Path-item `in: header` and `in: cookie` entries are now dropped rather than
inherited. Operation-level parameters are untouched, so an operation that
declares its own header keeps today's behavior.

Scope is now only what cannot be supplied any other way: `in: path` and
`in: query` parameters declared on the Path Item Object.

* fix(cli): merge allOf compositions in response schemas

Response-schema $refs were already expanded by deref_schema, but a schema
whose top level is an allOf composition was left unflattened. The properties
live inside the members, so a consumer reading them off the composition finds
none. fetcher_gen._data_schema looks for `properties.results`, gets nothing,
and falls through to _unwrap_schema_envelopes, which handles oneOf and anyOf
but not allOf. The generated Data model ends up with no fields.

Codat's listBills response is `allOf: [{properties: {results}}, {properties:
{_links, pageNumber, pageSize, totalResults}}]`, which produced an empty model.

merge_allof collapses a composition of object subschemas into one object
schema: properties unioned with the first declaration winning on conflict,
required unioned, and sibling keywords on the allOf node outranking members.
A composition that is not a plain intersection of object subschemas (a member
carrying its own oneOf/anyOf, or a non-object type) is returned unchanged
rather than flattened into something the spec does not describe. Nested
compositions are merged too, so a $ref inside a merged member still resolves.

Applied to extract_response_schema and extract_response_schemas.

Measured across five specs that use the pattern, counting operations whose
response schema exposes readable properties: 244 of 588 before, 394 after
(Codat Accounting 36 to 88, Codat Banking 0 to 8, Codat Commerce 1 to 11,
Rebilly 62 to 104, Box 145 to 183).

* fix(cli): merge allOf compositions in request body schemas

Same gap as the response side. request_body_parameters needs `type: object`
or `properties` to flatten a body into parameters, and an allOf composition
carries neither, so a POST or PUT generated from one gets no body parameters
at all.

extract_request_body_schema now runs merge_allof after deref_schema, matching
extract_response_schema.

Across the same five specs, counting request bodies that produce parameters:
138 of 206 before, 177 after (Codat Accounting 1 to 26 of its 30 bodies,
Rebilly 50 to 63, Box 87 to 88).

* fix(cli): stop annotation keywords blocking an allOf merge

_mergeable_allof_members required every member's keys to come from a small
whitelist, so any unrecognized keyword disqualified the whole composition.
Codat's PagingInfo carries `definitions` next to its properties, which meant
the live Codat-Accounting.yaml never merged and its response models stayed
empty. The APIs.guru copy of the same API omits that key, which is why the
earlier measurements looked clean.

The check is now a blocklist. A member blocks the merge when it carries its
own combinator (oneOf/anyOf/allOf/not), an unresolved $ref left by a cycle,
a scalar or array constraint (enum/const/items), or a type other than object.
Annotation keywords that do not affect the intersection (title, description,
examples, definitions, additionalProperties, vendor x- keys) no longer block
it, and are dropped from the merged result along with the members.

On the live Codat spec, operations whose response schema exposes readable
properties go from 36 of 127 to 114. Across the five specs measured earlier:
244 of 588 before, 488 after (was 394 under the whitelist).

Corpus check over 2,816 response and request-body extractions in 6 specs:
567 changed, every one attributable to an allOf in the pre-merge output,
0 unexplained. Twilio declares no allOf compositions and is unchanged.

* fix(cli): emit a name once when it is both a parameter and a body property

_signature_params appended request-body properties and then operation
parameters without deduplicating, so a name declared in both places produced
two function parameters with the same name and the generated module raised
SyntaxError.

Codat's push endpoints hit this: /push/bankAccounts/{accountId}/bankTransactions
declares accountId as a path parameter and repeats it in the request body, so
connections.push.bankAccounts.bankTransactions generated `accountId: str` and
`accountId: str = None` in one signature.

The operation parameter wins, since it carries the path placeholder and its own
required flag. The request payload is unchanged: _render_body_block derives body
fields from request_body_schema rather than from this list, so accountId is
still sent in the body.

This predates the path-item and allOf work in this PR. On origin/v5 the same
spec produces the same duplicate; on this branch the Codat spec generates 91
commands with 0 duplicated signatures, down from 1.

* fix(cli): use the safe identifier for keyword-named body properties

_render_body_block already writes `{'from': from_}`, keying the payload by the
wire name and reading a safe identifier, but _signature_params declared the raw
name. A body property named after a Python keyword therefore emitted
`from: str = None` and the generated module raised SyntaxError.

This did not fire before the allOf work in this PR, because a composed request
body produced no body parameters at all. Codat's Transfer body is such a
composition and has a `from` property, so connections.push.transfers started
generating invalid syntax once the body flattened.

_signature_params now emits safe_field_name(name)[0], the same helper
fetcher_gen and pydantic_gen already use. The body/query split in
_render_body_block compares against the safe names too: it previously matched
signature names against wire names, so `from_` was not recognized as a body
field and leaked into the query string as a `from_` parameter.

End to end on Codat-Accounting.yaml, generating the full extension and byte
compiling every emitted module: 136 files, 0 failures. origin/v5 emits 1 file
that does not compile (duplicate argument 'accountId').

Known gap left alone: an operation parameter named after a keyword would still
break, since the query and path lines use the raw name as a local variable.
None of the five specs checked declares one.

* fix(cli): do not treat instance data as schema, and sanitize POST identifiers

Two defects found by generating extensions from real specs and byte compiling
every emitted module.

merge_allof and deref_schema walked every value in a schema dict, including
keywords that hold instance data rather than subschemas. A `default` of
`{"allOf": [{"a": 1}, {"b": 2}]}` was rewritten to `{"type": "object"}`, and a
`$ref` key inside an example was resolved as if it were a reference. Both now
copy `default`, `example`, `examples`, `enum`, `const` and vendor `x-` values
through untouched.

_signature_params emitted operation parameter names verbatim, so a wire name
that is not a Python identifier produced invalid syntax. Rebilly declares an
`Organization-Id` header and a `REB-APIKEY` security header, which emitted
`REB-APIKEY: str = None`. It now uses safe_field_name, matching fetcher_gen,
and _render_body_block keys the query dict by the wire name while reading the
safe identifier so the request is unchanged.

This surfaced through the allOf work rather than being caused by it: a composed
request body used to yield no properties, so `has_body` was false in
package_gen and the command was generated as a GET fetcher, where identifiers
were already sanitized. With bodies flattening, 14 of Codat's 24 POST endpoints
correctly move to the POST path, which had never been exercised by these specs.

Generating the full extension and compiling every module:

  Codat-Accounting.yaml   v5: 1 failure    branch: 0   (136 files)
  Rebilly 2.1             v5: 14 failures  branch: 1   (97 files)

The one remaining Rebilly failure is `async def 3dsecure(` in the router, which
fails identically on v5. Sanitizing that is a naming decision about the command
surface, so it is left alone here.

Known gap: a non-credential header parameter is still placed in the query
string on the POST path. Pre-existing, and unchanged by this PR.

* fix(cli): resolve signature collisions on the emitted identifier

The previous dedupe compared wire names, but the failure it prevents depends on
the sanitized identifier. Two wire names can collapse onto one (`from` and
`from_`, `Organization-Id` and `Organization_Id`), and operation_parameters
keeps a name declared twice under different `in` locations by design, so both
cases could still emit a duplicate argument.

_signature_params now tracks the identifiers it has emitted and skips anything
that would repeat one. Emission order is unchanged, so existing signatures are
byte-identical; only genuinely colliding names are dropped, and operation
parameters still win over body properties.

Neither collision occurs in Codat-Accounting.yaml or Rebilly 2.1. This closes
the class rather than the instance, since the instances are what this PR has
been finding.

Generation unchanged: Codat 136 files 0 failures, Rebilly 97 files 1 failure
(`async def 3dsecure(`, identical on v5).

* test(cli): cover the instance-data guard, and drop a redundant branch

Mutation checking the new tests (revert each fix, confirm its test fails)
found two problems with the previous commits.

The guard that stops merge_allof and deref_schema from walking into `default`,
`example`, `enum`, `const` and vendor values shipped with no test at all. It is
now covered from both directions: a default holding an `allOf` key keeps its
value, and a `$ref` inside a default stays a literal while a real reference
beside it still resolves.

The body-property loop also carried a redundant check. It tested the raw wire
name against the operation parameter names before testing the sanitized one,
but `operation_param_names` holds both forms and `safe_field_name` is idempotent
on identifiers, so the first test could never fire alone. Reverting it left
every test passing, which is what surfaced it. Removed, and the sanitized name
is now computed once.

All nine mutations are caught by the suite: both data guards, the allOf
blocklist in each direction, the response merge wiring, body/parameter
precedence, the keyword identifier, the body/query split and the
emitted-identifier dedupe.

cli/tests at 1,954 passed against the same 80 environment failures as the base.
Generation unchanged: Codat 136 files 0 failures, Rebilly 97 files 1 failure.

* fix(cli): route required path-item headers/cookies through credential handling

* fix(cli): narrow Popen.stdin/stdout for ty in run_in_obb fixture

---------

Co-authored-by: deeleeramone <>

* fix warnings raised by federal_reserve on some xlsx workbook parsing

* fix publish workflows

* test publish pipeline with openbb-core-v2.0.0rc0 (#7681)

* fix publish workflows with missing tag job, some tests and bump fastapi

* fix python 3.10 test failure in core

* actually fix python 3.10 test failure in core

* deincrement openbb-core version to allow release pattern

* bump core version (#7682)

* remove uv.sources from pyproject.toml files and point to openbb-core-2.0.0 on PyPI

* remove oecd cache asset that wasn't supposed to be commuitted

* remove poetry.lock, update pyproject.tml and uv.lock for openbb-platform-api-v2.0.0 (#7684)

Co-authored-by: deeleeramone <>

* add the missing publish-openbb-platform-api.yml workflow file

* remove poetry.lock, update pyproject.tml and uv.lock for openbb-platform-api-v2.0.0 (#7685)

Co-authored-by: deeleeramone <>

* fix the tag generation after publish

* [Release] `openbb-charting-v4.0.0` (#7686)

* up openbb-charting to 4.0.0 and fill missing test coverage in chart_style

* test coverage and format markdown files

---------

Co-authored-by: deeleeramone <>

* remove scikit-learn from dependencies, update tests, and format markdown files (#7687)

Co-authored-by: deeleeramone <>

* [Release] `openbb-cli-v2.0.0` (#7692)

* fix missing test coverage in openbb_cli

* fix conftest

---------

Co-authored-by: deeleeramone <>

* add missing publish-openbb-cli.yml workflow

* [Release] `openbb-cli-v2.0.0` (#7693)

* fix missing test coverage in openbb_cli

* fix conftest

---------

Co-authored-by: deeleeramone <>

* test fixes for cli...

* [Release] `openbb-cli-v.2.0.0` (#7694)

* fix missing test coverage in openbb_cli

* fix conftest

---------

Co-authored-by: deeleeramone <>

* fix hanging mcp server test

* [Release] `openbb-mcp-server-v2.0.0` (#7695)

Co-authored-by: deeleeramone <>

* update uv.lock (#7696)

Co-authored-by: deeleeramone <>

* release openbb-commodity-v2.0.0 (#7697)

Co-authored-by: deeleeramone <>

* update uv.lock (#7698)

Co-authored-by: deeleeramone <>

* update uv.lock (#7699)

Co-authored-by: deeleeramone <>

* update uv.lock (#7700)

Co-authored-by: deeleeramone <>

* [Release] `openbb-econometrics-v2.0.0` (#7701)

* update uv.lock

* update pyproject.toml

---------

Co-authored-by: deeleeramone <>

* update uv.lock (#7703)

Co-authored-by: deeleeramone <>

* update uv.lock (#7702)

Co-authored-by: deeleeramone <>

* update uv.lock (#7704)

Co-authored-by: deeleeramone <>

* update uv.lock (#7705)

Co-authored-by: deeleeramone <>

* update uv.lock (#7706)

Co-authored-by: deeleeramone <>

* [V5] Fix a Number of V5 Bugs (#7707)

* fix a number of v5 bugs

* fix lint error

* more lint

* fixes for quantitative, technical, charting, famafrench for version bumps

---------

Co-authored-by: deeleeramone <>

* update uv.lock (#7709)

Co-authored-by: deeleeramone <>

* update uv.lock (#7708)

Co-authored-by: deeleeramone <>

* update uv.lock (#7710)

Co-authored-by: deeleeramone <>

* update uv.lock (#7711)

Co-authored-by: deeleeramone <>

* [Release] `openbb-cboe-v2.0.0` (#7713)

Co-authored-by: deeleeramone <>

* update uv.lock (#7714)

Co-authored-by: deeleeramone <>

* update uv.lock (#7717)

Co-authored-by: deeleeramone <>

* [Release] `openbb-ecb-v2.0.0` (#7715)

* update uv.lock

* update uv.lock

---------

Co-authored-by: deeleeramone <>

* update uv.lock (#7718)

Co-authored-by: deeleeramone <>

* update uv.lock (#7719)

Co-authored-by: deeleeramone <>

* update uv.lock (#7720)

Co-authored-by: deeleeramone <>

* add missing publish-openbb-fred.yml to repo

* update uv.lock (#7723)

Co-authored-by: deeleeramone <>

* update uv.lock (#7722)

Co-authored-by: deeleeramone <>

* update uv.lock (#7716)

Co-authored-by: deeleeramone <>

* update uv.lock (#7724)

Co-authored-by: deeleeramone <>

* update uv.lock (#7726)

Co-authored-by: deeleeramone <>

* [Release] `openbb-fred-v2.0.0` (#7725)

* update uv.lock

* fix tests and dependencies

* fix test_fred_fetchers

---------

Co-authored-by: deeleeramone <>

* [Release] `openbb-imf-v3.0.0` (#7727)

* update uv.lock

* fix tests and dependencies

* fix test_fred_fetchers

* bump openbb-imf version to 3.0.0

---------

Co-authored-by: deeleeramone <>

* update uv.lock (#7729)

Co-authored-by: deeleeramone <>

* update uv.lock (#7728)

Co-authored-by: deeleeramone <>

* update uv.lock (#7730)

Co-authored-by: deeleeramone <>

* update uv.lock (#7731)

Co-authored-by: deeleeramone <>

* update uv.lock (#7732)

Co-authored-by: deeleeramone <>

* update uv.lock (#7733)

Co-authored-by: deeleeramone <>

* cookiecutter pyproject and lock

* update pyproject.toml (#7734)

Co-authored-by: deeleeramone <>

* update pyproject.toml (#7735)

Co-authored-by: deeleeramone <>

* update uv.lock (#7736)

Co-authored-by: deeleeramone <>

* fix some sec issues, allegedly (#7737)

Co-authored-by: deeleeramone <>

* update uv.lock (#7738)

Co-authored-by: deeleeramone <>

* update main pyproject.toml

* lint fixes

* update assets (#7739)

Co-authored-by: deeleeramone <>

* [Release] `openbb-v5.0.0` (#7740)

* update assets

* add missing workflow

---------

Co-authored-by: deeleeramone <>

---------

Co-authored-by: Copilot <copilot@github.com>
Co-authored-by: deeleeramone <>
Co-authored-by: Theodore Aptekarev <aptekarev@gmail.com>
Co-authored-by: Dex <i@dex.moe>
Co-authored-by: Monte Hoover <monte.b.hoover@gmail.com>
Co-authored-by: Boris Li <65254053+BorisQuanLi@users.noreply.github.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
Co-authored-by: Lucas Zimmermann <lucaszmayer@gmail.com>
2026-09-29 14:45:53 +00:00

4220 lines
137 KiB
Python

"""Unit tests for ``openbb_sec.models.sec_financials``."""
import asyncio
import xml.etree.ElementTree as ET
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
import pydantic
import pytest
from openbb_sec.models import sec_financials as mod
from openbb_sec.models.sec_financials import (
FilingSectionQueryParams,
FinancialStatements,
_calendar_quarter,
_clean_disclosure_name,
_financial_statements_cache,
_is_amendment,
_is_annual_form,
_is_legal_redirect,
get_form10_urls_by_symbol,
no_filing_message,
resolve_filing_url,
resolve_section_url,
)
def make_fs(**attrs) -> FinancialStatements:
"""Build a FinancialStatements without running __init__, set PrivateAttrs."""
fs = object.__new__(FinancialStatements)
pydantic.BaseModel.__init__(fs)
for key, value in attrs.items():
setattr(fs, key, value)
return fs
def test_clean_disclosure_name_with_prefix():
assert (
_clean_disclosure_name("0007 - Disclosure - Inventory Disclosure")
== "Inventory Disclosure"
)
def test_clean_disclosure_name_no_prefix():
assert _clean_disclosure_name("Inventory Disclosure") == "Inventory Disclosure"
def test_calendar_quarter():
assert _calendar_quarter("2024-01-28") == 1
assert _calendar_quarter("2024-04-28") == 2
assert _calendar_quarter("2024-08-31") == 3
assert _calendar_quarter("2024-10-27") == 4
assert _calendar_quarter("") == 0
assert _calendar_quarter(None) == 0
assert _calendar_quarter("2024-XX-31") == 0
def test_is_amendment():
assert _is_amendment("10-K/A")
assert _is_amendment("20-F/A")
assert not _is_amendment("10-K")
assert not _is_amendment(None)
def test_is_annual_form():
assert _is_annual_form("10-K")
assert _is_annual_form("20-F")
# Amendments are distinct documents, not annual reports.
assert not _is_annual_form("40-F/A")
assert not _is_annual_form("10-K/A")
assert not _is_annual_form("10-Q")
assert not _is_annual_form("6-K")
assert not _is_annual_form(None)
def test_resolve_filing_url_prefers_original_over_amendment():
filings = [
{
"filing_type": "10-K/A",
"period_ending": "1997-09-26",
"filing_date": "1998-01-23",
"url": "amendment",
},
{
"filing_type": "10-K",
"period_ending": "1997-09-26",
"filing_date": "1997-12-05",
"url": "original",
},
]
with patch.object(
mod, "get_form10_urls_by_symbol", new=AsyncMock(return_value=filings)
):
url = asyncio.run(
resolve_filing_url("AAPL", calendar_year=1997, annual_default=True)
)
assert url == "original"
def test_no_filing_message():
msg = no_filing_message("BABA")
assert "BABA" in msg
assert "20-F" in msg and "foreign private issuer" in msg
def test_is_legal_redirect():
assert _is_legal_redirect(
"Please see Note 12 of the Notes to the Consolidated Financial Statements."
)
assert _is_legal_redirect("This matter is incorporated herein by reference.")
assert not _is_legal_redirect("We are a defendant in a patent suit filed in 2023.")
assert not _is_legal_redirect("x" * 1600)
def test_filing_section_query_params_validators():
q = FilingSectionQueryParams(symbol="wdfc", calendar_year="", calendar_period="") # ty: ignore[invalid-argument-type]
assert q.symbol == "WDFC"
assert q.calendar_year is None
assert q.calendar_period is None
def test_filing_section_query_params_to_upper_non_string():
assert FilingSectionQueryParams._to_upper(123) == 123
def test_filing_section_query_params_empty_to_none_passthrough():
assert FilingSectionQueryParams._empty_to_none("q1") == "q1"
def test_resolve_section_url_explicit_url():
q = SimpleNamespace(url="https://sec.gov/x", symbol="WDFC", use_cache=True)
assert asyncio.run(resolve_section_url(q)) == "https://sec.gov/x"
def test_resolve_section_url_no_symbol():
q = SimpleNamespace(url=None, symbol=None, use_cache=True)
assert asyncio.run(resolve_section_url(q)) == ""
def test_resolve_section_url_from_symbol():
q = SimpleNamespace(
url=None,
symbol="WDFC",
calendar_year=2024,
calendar_period="Q1",
use_cache=False,
)
with patch.object(
mod, "resolve_filing_url", new=AsyncMock(return_value="resolved")
) as rfu:
assert asyncio.run(resolve_section_url(q)) == "resolved"
rfu.assert_awaited_once_with("WDFC", 2024, "Q1", False, True)
def test_resolve_section_url_passes_annual_default():
q = SimpleNamespace(
url=None,
symbol="X",
calendar_year=None,
calendar_period=None,
use_cache=False,
)
with patch.object(
mod, "resolve_filing_url", new=AsyncMock(return_value="u")
) as rfu:
asyncio.run(resolve_section_url(q, annual_default=False))
rfu.assert_awaited_once_with("X", None, None, False, False)
def test_resolve_filing_url_latest_filing_any_form():
filings = [
{
"filing_date": "2024-10-01",
"period_ending": "2024-08-31",
"filing_type": "10-K",
"url": "k2024",
},
{
"filing_date": "2024-12-15",
"period_ending": "2024-11-30",
"filing_type": "10-Q",
"url": "q-latest",
},
]
with patch.object(
mod, "get_form10_urls_by_symbol", new=AsyncMock(return_value=filings)
):
assert asyncio.run(resolve_filing_url("X")) == "k2024"
assert asyncio.run(resolve_filing_url("X", annual_default=False)) == "q-latest"
def test_get_form10_urls_by_symbol():
filing = SimpleNamespace(
filing_date="2024-10-01",
report_date="2024-08-31",
report_type="10-K",
report_url="https://sec.gov/k",
)
class _Fetcher:
async def fetch_data(self, params, info):
assert params["form_type"] == "10-K,10-Q,20-F,40-F"
return [filing]
with patch(
"openbb_sec.models.company_filings.SecCompanyFilingsFetcher",
_Fetcher,
):
result = asyncio.run(get_form10_urls_by_symbol("WDFC", use_cache=False))
assert result == [
{
"filing_date": "2024-10-01",
"period_ending": "2024-08-31",
"filing_type": "10-K",
"url": "https://sec.gov/k",
}
]
def _filings():
return [
{
"filing_date": "2024-10-01",
"period_ending": "2024-08-31",
"filing_type": "10-K",
"url": "k2024",
},
{
"filing_date": "2023-10-01",
"period_ending": "2023-08-31",
"filing_type": "10-K",
"url": "k2023",
},
{
"filing_date": "2024-04-01",
"period_ending": "2024-02-28",
"filing_type": "10-Q",
"url": "q1-2024",
},
{
"filing_date": "2024-07-01",
"period_ending": "2024-05-31",
"filing_type": "10-Q",
"url": "q2-2024",
},
]
def test_resolve_filing_url_annual_latest():
with patch.object(
mod, "get_form10_urls_by_symbol", new=AsyncMock(return_value=_filings())
):
assert asyncio.run(resolve_filing_url("WDFC")) == "k2024"
def test_resolve_filing_url_annual_by_year():
with patch.object(
mod, "get_form10_urls_by_symbol", new=AsyncMock(return_value=_filings())
):
assert asyncio.run(resolve_filing_url("WDFC", calendar_year=2023)) == "k2023"
def test_resolve_filing_url_by_calendar_period():
with patch.object(
mod, "get_form10_urls_by_symbol", new=AsyncMock(return_value=_filings())
):
# 2024-02-28 is calendar Q1; 2024-05-31 is calendar Q2.
assert (
asyncio.run(resolve_filing_url("WDFC", calendar_period="Q1")) == "q1-2024"
)
assert (
asyncio.run(resolve_filing_url("WDFC", calendar_period="Q2")) == "q2-2024"
)
def test_resolve_filing_url_none_found():
with patch.object(
mod, "get_form10_urls_by_symbol", new=AsyncMock(return_value=_filings())
):
assert asyncio.run(resolve_filing_url("WDFC", calendar_year=1999)) == ""
def test_init_non_xbrl_branch():
fs = object.__new__(FinancialStatements)
pydantic.BaseModel.__init__(fs)
with (
patch("openbb_sec.models.sec_filing.Filing.__init__", return_value=None),
patch.object(FinancialStatements, "_download_metalinks"),
patch.object(FinancialStatements, "_initialize_non_xbrl") as init_non,
patch.object(FinancialStatements, "_build_schema_from_xml") as build,
):
FinancialStatements.__init__(fs, "https://www.sec.gov/x")
init_non.assert_called_once()
build.assert_not_called()
def test_init_xbrl_branch():
"""An XBRL filing skips non-XBRL init and defers schema building."""
fs = object.__new__(FinancialStatements)
pydantic.BaseModel.__init__(fs)
fs._resources = {"r1": {}, "r2": {}}
with (
patch("openbb_sec.models.sec_filing.Filing.__init__", return_value=None),
patch.object(FinancialStatements, "_download_metalinks"),
patch.object(FinancialStatements, "_build_schema_from_xml") as build,
patch.object(FinancialStatements, "_download_xbrl_instance"),
patch.object(FinancialStatements, "_enrich_cover_page_from_instance"),
patch.object(FinancialStatements, "_initialize_non_xbrl") as init_non,
):
FinancialStatements.__init__(fs, "https://www.sec.gov/x")
build.assert_not_called()
init_non.assert_not_called()
def test_ensure_xbrl_builds_once():
"""The first XBRL access builds schema and instance; later calls are no-ops."""
fs = object.__new__(FinancialStatements)
pydantic.BaseModel.__init__(fs)
fs._resources = {"r1": {}, "r2": {}}
fs._xbrl_loaded = False
with (
patch.object(FinancialStatements, "_build_schema_from_xml") as build,
patch.object(FinancialStatements, "_download_xbrl_instance") as inst,
patch.object(FinancialStatements, "_enrich_cover_page_from_instance") as enrich,
):
fs._ensure_xbrl()
fs._ensure_xbrl()
build.assert_called_once()
inst.assert_called_once()
enrich.assert_called_once()
def test_ensure_xbrl_skips_when_not_xbrl():
"""A non-XBRL filing marks itself loaded without building anything."""
fs = object.__new__(FinancialStatements)
pydantic.BaseModel.__init__(fs)
fs._resources = {}
fs._xbrl_loaded = False
with patch.object(FinancialStatements, "_build_schema_from_xml") as build:
fs._ensure_xbrl()
assert fs._xbrl_loaded is True
build.assert_not_called()
def test_from_url_memoized():
_financial_statements_cache.cache_clear()
sentinel = object()
with patch.object(mod, "FinancialStatements", return_value=sentinel) as ctor:
a = FinancialStatements.from_url("u", True)
b = FinancialStatements.from_url("u", True)
assert a is sentinel and b is sentinel
ctor.assert_called_once_with("u", True)
_financial_statements_cache.cache_clear()
def test_get_item_sections_cached():
fs = make_fs(_items={"item_1": {"name": "Business"}})
assert fs.get_item_sections() == {"item_1": {"name": "Business"}}
def test_get_item_sections_no_content():
fs = make_fs()
with patch.object(fs, "get_main_document_content", return_value=None):
assert fs.get_item_sections() == {}
def test_get_item_sections_builds_from_html():
fs = make_fs()
with (
patch.object(fs, "get_main_document_content", return_value="<html>x</html>"),
patch.object(
fs, "_clean_html_to_text", return_value="## Item 1\nBusiness\nText."
),
patch(
"openbb_sec.utils.filing_sections.extract_item_sections",
return_value={"item_1": {"name": "Business", "text": "Text."}},
),
):
out = fs.get_item_sections()
assert out == {"item_1": {"name": "Business", "text": "Text."}}
assert fs._items == out
def test_get_item_sections_raw_fallback():
# Old plain-text filings: markdown finds nothing, so items come from raw text.
fs = make_fs()
content = "Item 1. Legal Proceedings\n\nReference is made to the 10-K."
with (
patch.object(fs, "get_main_document_content", return_value=content),
patch.object(fs, "_clean_html_to_text", return_value="prose with no headers"),
):
items = fs.get_item_sections()
assert "item_1" in items
assert items["item_1"]["name"] == "Legal Proceedings"
def test_get_item_matches_item_num():
fs = make_fs(_items={"x": {"item_num": "1A", "name": "Risk"}})
assert fs.get_item("1a") == {"item_num": "1A", "name": "Risk"}
def test_get_item_matches_key_suffix():
fs = make_fs(_items={"item_II_1": {"name": "Legal"}})
assert fs.get_item("1") == {"name": "Legal"}
def test_get_item_no_match():
fs = make_fs(_items={"item_1": {"item_num": "1", "name": "Biz"}})
assert fs.get_item("99") is None
def test_item_by_name():
fs = make_fs(_items={"k": {"name": "The Business Overview"}})
assert fs._item_by_name("business") == {"name": "The Business Overview"}
assert fs._item_by_name("missing") is None
def test_business_10k_path():
fs = make_fs(
_document_type="10-K", _items={"item_1": {"item_num": "1", "text": " Body "}}
)
assert fs.business() == "Body"
def test_business_non_10k_by_name():
fs = make_fs(
_document_type="10-Q", _items={"x": {"name": "Our Business", "text": "Body"}}
)
assert fs.business() == "Body"
def test_business_foreign_slices_section_from_extracted_html():
fs = make_fs(
_document_type="20-F",
_items={"x": {"item_num": "4", "name": "Info", "text": "fallback"}},
)
section_text = "Intro\nB. Business\nOperating discussion\nITEM 5\nOther"
with (
patch.object(fs, "get_main_document_content", return_value="<html/>"),
patch(
"openbb_sec.utils.filing_sections.extract_section_html",
return_value="<div>section</div>",
),
patch.object(fs, "_clean_html_to_text", return_value=section_text),
):
out = fs.business()
assert out == "B. Business\nOperating discussion"
def test_business_foreign_fallback_text_without_business_heading():
fs = make_fs(
_document_type="20-F",
_items={"x": {"item_num": "4", "name": "Info", "text": "General company text"}},
)
with patch.object(fs, "get_main_document_content", return_value=None):
out = fs.business()
assert out == "General company text"
def test_business_foreign_falls_back_to_business_name_lookup():
fs = make_fs(
_document_type="20-F",
_items={"x": {"name": "Our business", "text": "Body"}},
)
assert fs.business() == "Body"
def test_business_none_when_missing():
fs = make_fs(_document_type="10-K", _items={"item_2": {"item_num": "2"}})
assert fs.business() is None
def test_business_none_when_empty_text():
fs = make_fs(
_document_type="10-K", _items={"item_1": {"item_num": "1", "text": " "}}
)
assert fs.business() is None
def test_legal_proceedings_by_name():
fs = make_fs(_items={"k": {"name": "Legal Proceedings", "text": "Suit"}})
assert fs.legal_proceedings() == {"name": "Legal Proceedings", "text": "Suit"}
def test_legal_proceedings_by_item():
fs = make_fs(_items={"item_3": {"item_num": "3", "name": "Other"}})
assert fs.legal_proceedings() == {"item_num": "3", "name": "Other"}
def test_legal_proceedings_no_item():
assert make_fs(_items={"item_2": {"item_num": "2"}}).legal_proceedings() is None
def test_legal_proceedings_follows_redirect():
fs = make_fs(
_items={
"k": {
"name": "Legal Proceedings",
"text": "Please see Note 12 of the Notes to the Consolidated "
"Financial Statements.",
}
},
_disclosures={
"g": {
"name": "Commitments and Contingencies",
"text": "The Company is subject to various legal proceedings.",
}
},
)
result = fs.legal_proceedings()
assert "Please see Note 12" in result["text"]
assert "subject to various legal proceedings" in result["text"]
def test_legal_proceedings_redirect_without_note():
fs = make_fs(
_items={
"k": {
"name": "Legal Proceedings",
"text": "Please see Note 12 of the Notes to the Financial Statements.",
}
},
_disclosures={},
)
assert fs.legal_proceedings()["text"].startswith("Please see Note 12")
def test_legal_contingencies_note_selection():
fs = make_fs(
_disclosures={
"Inventory": {"name": "Inventory Disclosure", "text": "inv"},
"contingencies_raw": "raw",
"Contingencies (Details)": {
"name": "Contingencies (Details)",
"text": "y",
},
"legal": {"name": "Legal Matters", "text": ""},
"good": {"name": "Commitments and Contingencies", "text": "the note"},
}
)
assert fs._legal_contingencies_note() == "the note"
def test_legal_contingencies_note_none_match():
fs = make_fs(_disclosures={"a": {"name": "Inventory Disclosure", "text": "x"}})
assert fs._legal_contingencies_note() == ""
def test_risk_factors_no_item():
fs = make_fs(_items={"item_2": {"item_num": "2"}})
assert fs.risk_factors() == []
def test_risk_factors_foreign_item_3d_path():
fs = make_fs(
_document_type="20-F", _items={"x": {"item_num": "3D", "text": " body "}}
)
with patch.object(fs, "get_main_document_content", return_value=None):
out = fs.risk_factors()
assert out == [{"risk_factor": None, "text": "body"}]
def test_risk_factors_blocks_from_html():
fs = make_fs(_items={"x": {"name": "Risk Factors", "item_num": "1A"}})
# The headingless preamble block (risk_factor=None) is dropped.
blocks = [{"risk_factor": None, "text": "intro"}, {"risk_factor": "A", "text": "t"}]
with (
patch.object(fs, "get_main_document_content", return_value="<html>sec</html>"),
patch(
"openbb_sec.utils.filing_sections.extract_section_html",
return_value="<p>section</p>",
),
patch(
"openbb_sec.utils.filing_sections.split_bold_sections", return_value=blocks
),
):
assert fs.risk_factors() == [{"risk_factor": "A", "text": "t"}]
def test_risk_factors_fallback_text():
fs = make_fs(
_items={"x": {"name": "Risk Factors", "item_num": "1A", "text": " body "}}
)
with (
patch.object(fs, "get_main_document_content", return_value="<html>sec</html>"),
patch("openbb_sec.utils.filing_sections.extract_section_html", return_value=""),
):
assert fs.risk_factors() == [{"risk_factor": "Risk Factors", "text": "body"}]
def test_risk_factors_no_content_no_text():
fs = make_fs(_items={"x": {"name": "Risk Factors", "item_num": "1A", "text": ""}})
with patch.object(fs, "get_main_document_content", return_value=None):
assert fs.risk_factors() == []
def test_segment_revenue_collects_matching():
fs = make_fs(
_disclosures={
"seg": {
"name": "0001 - Disclosure - Segment Information",
"text": " data ",
},
"tab": {"name": "Segment Information (Tables)", "text": "skip"},
"other": {"name": "Inventory", "text": "x"},
}
)
out = fs.segment_revenue()
assert out == [{"name": "Segment Information", "text": "data"}]
def test_segment_revenue_skips_empty_text():
fs = make_fs(
_disclosures={"seg": {"name": "Revenue Disaggregation", "text": " "}}
)
assert fs.segment_revenue() == []
def test_calendar_period():
fs = make_fs(_period_ending="2024-08-31")
assert fs.calendar_period == "2024-Q3"
def test_fiscal_period_variants():
assert make_fs(_fiscal_year="2024", _fiscal_period="FY").fiscal_period == "2024-FY"
assert make_fs(_fiscal_year="", _fiscal_period="Q1").fiscal_period == "Q1"
assert make_fs(_fiscal_year="2024", _fiscal_period="").fiscal_period == "2024"
assert make_fs(_fiscal_year="", _fiscal_period="").fiscal_period == ""
def test_is_xbrl_true_and_false():
assert make_fs(_metalinks={"a": 1}).is_xbrl is True
assert make_fs(_instance={"a": 1}).is_xbrl is True
assert make_fs().is_xbrl is False
def test_main_document_url_and_get():
fs = make_fs(
_document_urls=[
{"type": "EX-101.INS", "url": "ins.xml"},
{"type": "10-K", "url": "main.htm"},
]
)
assert fs.main_document_url == "main.htm"
def test_get_main_document_url_html_fallback():
fs = make_fs(_document_urls=[{"type": "GRAPHIC", "url": "x.htm"}])
assert fs.get_main_document_url() == "x.htm"
def test_get_main_document_url_none():
assert make_fs().get_main_document_url() is None
fs = make_fs(_document_urls=[{"type": "GRAPHIC", "url": "x.jpg"}])
assert fs.get_main_document_url() is None
def test_statements_xbrl():
fs = make_fs(
_resources={
"r1": {
"group": "statement",
"short_name": "Income",
"long_name": "Income Statement",
"url": "u",
},
"r2": {"group": "disclosure", "short_name": "Note"},
}
)
st = fs.statements
assert list(st.keys()) == ["Income"]
assert st["Income"] == {"name": "Income Statement", "url": "u"}
def test_statements_non_xbrl():
fs = make_fs(
_non_xbrl_statements={"income": {"name": "Income", "data": "D", "meta": "M"}}
)
st = fs.statements
assert st["income"] == {"name": "Income", "data": "D", "meta": "M"}
def test_statements_empty():
assert dict(make_fs().statements.labels()) == {}
def test_get_statements_info_xbrl_no_statement_group():
fs = make_fs(_resources={"r1": {"group": "disclosure", "short_name": "Note"}})
assert fs._get_statements_info() == {}
def test_disclosures_non_xbrl_fallback():
fs = make_fs(_disclosures={"d1": {"name": "Note 1", "text": "t"}})
assert fs.disclosures["d1"] == {"name": "Note 1", "text": "t"}
def test_disclosures_items_fallback_and_label():
fs = make_fs(_items={"item_1": "raw string value"})
d = fs.disclosures
assert d.labels()["item_1"] == "item_1"
assert d["item_1"] == "raw string value"
def test_disclosures_empty():
assert dict(make_fs().disclosures.labels()) == {}
def test_disclosures_xbrl_direct_key_match():
fs = make_fs(
_resources={
"us-gaap_InventoryDisclosureTextBlock": {
"group": "disclosure",
"name": "us-gaap_InventoryDisclosureTextBlock",
"long_name": "Inventory",
"short_name": "Inventory",
}
},
_text_blocks={
"us-gaap_InventoryDisclosureTextBlock": {
"value": "Inventory text",
"presentation": [],
}
},
)
info = fs._get_disclosures_info()
assert info["us-gaap_InventoryDisclosureTextBlock"]["text"] == "Inventory text"
def test_disclosures_xbrl_role_exact_match_and_append():
fs = make_fs(
_resources={
"us-gaap_SegmentDisclosureTextBlock": {
"group": "disclosure",
"anchor_tag": "us-gaap_SegmentDisclosureTextBlock",
"long_name": "Segment",
"short_name": "Segment",
}
},
_text_blocks={
"tb1": {"value": "first", "presentation": ["SegmentDisclosureTextBlock"]},
"tb2": {"value": "second", "presentation": ["SegmentDisclosureTextBlock"]},
},
)
info = fs._get_disclosures_info()
text = info["us-gaap_SegmentDisclosureTextBlock"]["text"]
assert "first" in text and "second" in text
def test_disclosures_xbrl_partial_role_match():
fs = make_fs(
_resources={
"us-gaap_RevenueTextBlock": {
"group": "disclosure",
"anchor_tag": "us-gaap_RevenueTextBlock",
"long_name": "Revenue",
"short_name": "Revenue",
}
},
_text_blocks={
"tb1": {"value": "rev", "presentation": ["RevenueTextBlockExtra"]},
},
)
info = fs._get_disclosures_info()
assert info["us-gaap_RevenueTextBlock"]["text"] == "rev"
def test_disclosures_xbrl_local_name_fallback_match():
fs = make_fs(
_resources={
"us-gaap_LeasesDisclosure": {
"group": "disclosure",
"anchor_tag": "us-gaap_LeasesDisclosure",
"long_name": "Leases",
"short_name": "Leases",
}
},
_text_blocks={
"srt_LeasesDisclosure": {
"value": "leases text",
"presentation": ["Unrelated"],
},
},
)
info = fs._get_disclosures_info()
assert info["us-gaap_LeasesDisclosure"]["text"] == "leases text"
def test_disclosures_xbrl_skips_none_textblock():
fs = make_fs(
_resources={
"us-gaap_X": {
"group": "disclosure",
"anchor_tag": "us-gaap_X",
"long_name": "X",
"short_name": "X",
}
},
_text_blocks={"tb": {"value": "\nNone.", "presentation": ["X"]}},
)
info = fs._get_disclosures_info()
assert info["us-gaap_X"]["text"] == ""
def test_text_blocks_xbrl():
fs = make_fs(
_resources={
"us-gaap_InventoryDisclosureTextBlock": {
"group": "disclosure",
"anchor_tag": "us-gaap_InventoryDisclosureTextBlock",
"long_name": "Inventory",
"short_name": "Inventory",
}
},
_text_blocks={
"us-gaap_InventoryDisclosureTextBlock": {
"value": "Inventory text",
"presentation": ["InventoryDisclosure"],
"name": "InventoryDisclosure",
},
"skip": {"value": "\nNone.", "presentation": []},
},
)
tb = fs.text_blocks
assert "us-gaap_InventoryDisclosureTextBlock" in tb
assert tb["us-gaap_InventoryDisclosureTextBlock"]["text"] == "Inventory text"
assert "skip" not in tb
def test_text_blocks_non_xbrl_and_empty():
fs = make_fs(_non_xbrl_text_blocks={"a": {"text": "x"}})
assert fs.text_blocks == {"a": {"text": "x"}}
assert make_fs().text_blocks == {}
def test_tags_property():
fs = make_fs(_tags={"a": {"name": "A"}})
assert fs.tags == {"a": {"name": "A"}}
def test_toc_cached():
fs = make_fs(_toc={"1A": "Risk Factors"})
assert fs.toc == {"1A": "Risk Factors"}
def test_toc_no_content():
fs = make_fs()
with patch.object(fs, "get_main_document_content", return_value=None):
assert fs.toc == {}
def test_toc_builds_from_links():
html = (
'<a href="#item1a">Item 1A</a>'
'<a href="#item1a">Risk Factors and uncertainties</a>'
'<a href="#page">5</a>'
'<a href="">empty</a>'
'<a href="http://x">external</a>'
'<a href="#x"></a>'
)
fs = make_fs()
with patch.object(fs, "get_main_document_content", return_value=html):
toc = fs.toc
assert toc == {"1A": "Risk Factors and uncertainties"}
def test_get_statement_delegates():
fs = make_fs()
with patch.object(fs, "_download_statement", return_value=("d", "m")) as ds:
assert fs.get_statement("balance") == ("d", "m")
ds.assert_called_once_with("balance")
def test_download_statement_no_matched():
fs = make_fs(
_instance={"x": 1},
_resources={
"r1": {"group": "statement", "short_name": "Balance Sheet", "url": "u"}
},
)
with pytest.raises(ValueError, match="No items found"):
fs._download_statement("income")
def test_download_statement_no_urls():
fs = make_fs(
_instance={"x": 1},
_resources={
"r1": {"group": "statement", "short_name": "Income Statement", "url": None}
},
)
with pytest.raises(ValueError, match="No URLs found"):
fs._download_statement("income")
def test_download_statement_cash_search_term_no_match():
fs = make_fs(
_instance={"x": 1},
_resources={
"r1": {"group": "statement", "short_name": "Income Statement", "url": "u"}
},
)
with pytest.raises(ValueError, match="No items found"):
fs._download_statement("cash")
def _income_fs():
from pandas import DataFrame
fs = make_fs(
_document_type="10-K",
_period_ending="2024-08-31",
_instance={
"us-gaap_Revenues": {
"context": [
{
"value": "100",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "iso4217:USD",
"decimals": "-6",
},
{
"value": "90",
"context_ref": "duration_2022_09_01_to_2023_08_31",
"unit": "iso4217:USD",
"decimals": "-6",
},
]
},
"us-gaap_NetIncomeLoss": {
"context": [
{
"value": "50",
"context_ref": "as_of_2024_08_31",
"unit": "iso4217:USD",
"decimals": "-6",
}
]
},
},
_tags={
"us-gaap_Revenues": {
"label": "Net sales and revenues",
"terseLabel": "Net sales and revenues",
"crdr": "credit",
"weight": "1",
"parent_tag": "us-gaap_IncomeStatement",
"preferred_label": "totalLabel",
"name": "Revenues",
"period_type": "duration",
},
"us-gaap_NetIncomeLoss": {
"label": "Net income",
"crdr": "credit",
"name": "NetIncomeLoss",
"period_type": "duration",
},
},
_period_context={
"duration_2023_09_01_to_2024_08_31": {"start": "2023-09-01"},
},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
table = DataFrame(
{
"Statement of Operations $ in Millions": [
"Net sales and revenues",
"Net income",
],
"2024-08-31 -- 12 Months Ended": [100, 50],
"2023-08-31 -- 12 Months Ended": [90, 40],
}
)
meta = DataFrame(
{
"Name": ["us-gaap_Revenues"],
"Namespace Prefix": ["us-gaap"],
"Data Type": ["xbrli:monetaryItemType"],
"Balance Type": ["credit"],
"Period Type": ["duration"],
"unit": ["iso4217:USD"],
"decimals": ["-6"],
"weight": ["1"],
"preferred_label": ["totalLabel"],
"parent_tag": ["us-gaap_IncomeStatement"],
}
)
return fs, table, meta
def test_download_statement_income_pipeline():
fs, table, meta = _income_fs()
with patch.object(fs, "_download_statement_from_url", return_value=(table, meta)):
data, out_meta = fs._download_statement("income")
assert "tag" in data.columns
assert "period_ending" in data.columns
assert not data.empty
assert set(["us-gaap_Revenues"]).issubset(set(data.tag.dropna().tolist()))
def test_download_statement_download_loop_raises():
fs, _, _ = _income_fs()
with patch.object(
fs, "_download_statement_from_url", side_effect=RuntimeError("boom")
):
with pytest.raises(RuntimeError, match="Failed to download statement"):
fs._download_statement("income")
def test_download_statement_balance_fix_tag_and_dimensions():
from pandas import DataFrame
fs = make_fs(
_document_type="10-K",
_period_ending="2024-08-31",
_instance={
"us-gaap_Assets": {
"context": [
{
"value": "200",
"context_ref": "as_of_2024_08_31",
"unit": "unit_Standard_iso4217_USD",
"decimals": "-6",
}
]
},
"us-gaap_AssetsCurrent": {
"context": [{"value": "200", "context_ref": "as_of_2024_08_31"}]
},
"us-gaap_Revenues": {
"context": [
{
"value": "75",
"context_ref": "duration_2023_09_01_to_2024_08_31_us-gaap_ProductMember",
}
]
},
},
_tags={
"us-gaap_Assets": {"label": "Total assets", "crdr": "debit"},
"us-gaap_AssetsCurrent": {"label": "Current assets", "crdr": "debit"},
"us-gaap_Wrong": {"label": "Total assets", "crdr": "debit"},
"us-gaap_Revenues": {"label": "Revenue", "crdr": "credit"},
"us-gaap_ProductMember": {"terseLabel": "Product line"},
},
_period_context={},
_resources={
"r1": {
"group": "statement",
"short_name": "Balance Sheet",
"url": "https://sec.gov/R2.htm",
}
},
)
table = DataFrame(
{
"Balance Sheet $ in Millions": ["Total assets", "Revenue"],
"2024-08-31": [200, 75],
}
)
meta = DataFrame()
with patch.object(fs, "_download_statement_from_url", return_value=(table, meta)):
data, _ = fs._download_statement("balance")
labels = data.label.tolist()
assert "Product line" in labels or "Revenue" in labels
assert not data.empty
def test_download_statement_equity_pipeline():
from pandas import DataFrame
fs = make_fs(
_document_type="10-K",
_instance={
"us-gaap_StockholdersEquity": {
"context": [
{
"value": "500",
"context_ref": "as_of_2024_08_31",
"unit": "iso4217:USD",
}
]
}
},
_tags={
"us-gaap_StockholdersEquity": {
"label": "Total stockholders' equity",
"crdr": "credit",
}
},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Equity",
"url": "https://sec.gov/R7.htm",
}
},
)
table = DataFrame(
{
"Statement of Equity": ["Total stockholders' equity"],
"Common Stock": [500],
}
)
meta = DataFrame()
with patch.object(fs, "_download_statement_from_url", return_value=(table, meta)):
data, out_meta = fs._download_statement("equity")
assert "tag" in data.columns
assert data.tag.iloc[0] == "us-gaap_StockholdersEquity"
def test_disclosures_partial_role_append_existing():
fs = make_fs(
_resources={
"us-gaap_DebtTextBlock": {
"group": "disclosure",
"anchor_tag": "us-gaap_DebtTextBlock",
"long_name": "Debt",
"short_name": "Debt",
}
},
_text_blocks={
"tb1": {"value": "first", "presentation": ["DebtTextBlock"]},
"tb2": {"value": "second", "presentation": ["DebtTextBlockMore"]},
},
)
info = fs._get_disclosures_info()
text = info["us-gaap_DebtTextBlock"]["text"]
assert "first" in text and "second" in text
def test_disclosures_local_fallback_append_existing():
fs = make_fs(
_resources={
"us-gaap_TaxTextBlock": {
"group": "disclosure",
"anchor_tag": "us-gaap_TaxTextBlock",
"long_name": "Tax",
"short_name": "Tax",
}
},
_text_blocks={
"us-gaap_TaxTextBlock": {"value": "first", "presentation": []},
"us-gaap_TaxTextBlockExtra": {"value": "second", "presentation": []},
},
)
info = fs._get_disclosures_info()
text = info["us-gaap_TaxTextBlock"]["text"]
assert "first" in text and "second" in text
def test_disclosures_resource_without_tag_skipped():
fs = make_fs(_resources={"r": {"group": "disclosure"}})
assert fs._get_disclosures_info() == {}
def test_text_blocks_exact_role_match():
fs = make_fs(
_resources={
"us-gaap_InventoryDisclosure": {
"group": "disclosure",
"anchor_tag": "us-gaap_InventoryDisclosure",
"long_name": "Inventory",
"short_name": "Inventory",
}
},
_text_blocks={
"tb1": {"value": "inv text", "presentation": ["InventoryDisclosure"]},
},
)
tb = fs.text_blocks
assert tb["tb1"]["disclosure"] == ["us-gaap_InventoryDisclosure"]
def test_text_blocks_partial_role_match():
fs = make_fs(
_resources={
"us-gaap_DebtDisclosure": {
"group": "disclosure",
"anchor_tag": "us-gaap_DebtDisclosure",
"long_name": "Debt",
"short_name": "Debt",
}
},
_text_blocks={
"tb1": {"value": "debt text", "presentation": ["DebtDisclosureExtra"]},
},
)
tb = fs.text_blocks
assert tb["tb1"]["disclosure"] == ["us-gaap_DebtDisclosure"]
def test_enrich_cover_page_no_instance():
fs = make_fs()
fs._enrich_cover_page_from_instance()
assert fs._cover_page == {}
def test_enrich_cover_page_populates_fields():
fs = make_fs(
_period_ending="2024-08-31",
_instance={
"dei_TradingSymbol": {"context": [{"value": "WDFC", "end": "2024-08-31"}]},
"dei_EntityCommonStockSharesOutstanding": {
"context": [{"value": "13500000", "end": "2024-09-30"}]
},
"dei_DocumentFiscalYearFocus": {
"context": [{"value": "2024", "end": "2024-08-31"}]
},
"dei_DocumentFiscalPeriodFocus": {
"context": [{"value": "FY", "end": "2024-08-31"}]
},
},
)
fs._enrich_cover_page_from_instance()
assert "WDFC" in fs._trading_symbols
assert fs._shares_outstanding == {"2024-09-30": 13500000}
assert fs._fiscal_year == "2024"
assert fs._fiscal_period == "FY"
def test_enrich_cover_page_skips_already_set_and_dimensional_and_empty():
fs = make_fs(
_cover_page={"Trading Symbol": "EXISTING"},
_instance={
"dei_TradingSymbol": {"context": [{"value": "NEW", "end": "2024-08-31"}]},
"dei_DocumentFiscalYearFocus": {"context": []},
"dei_DocumentFiscalPeriodFocus": {
"context": [{"value": "0"}, {"dimensions": {"x": 1}, "value": "Q1"}]
},
"Unknown": {"context": [{"value": "x"}]},
},
)
fs._enrich_cover_page_from_instance()
assert fs._cover_page["Trading Symbol"] == "EXISTING"
assert fs._fiscal_period == ""
def test_enrich_cover_page_shares_value_error():
fs = make_fs(
_instance={
"dei_EntityCommonStockSharesOutstanding": {
"context": [{"value": "not-a-number", "end": "2024-08-31"}]
}
}
)
fs._enrich_cover_page_from_instance()
assert fs._shares_outstanding == {}
def test_parse_non_xbrl_statement_no_content():
fs = make_fs()
with patch.object(fs, "get_main_document_content", return_value=None):
with pytest.raises(ValueError, match="No main document content"):
fs._parse_non_xbrl_statement("income")
def test_parse_non_xbrl_statement_empty_df():
from pandas import DataFrame
fs = make_fs()
with (
patch.object(fs, "get_main_document_content", return_value="<html></html>"),
patch(
"openbb_sec.utils.non_xbrl_parser.parse_non_xbrl_statement",
return_value=(DataFrame(), DataFrame()),
),
):
with pytest.raises(ValueError, match="Could not find income statement"):
fs._parse_non_xbrl_statement("income")
def test_parse_non_xbrl_statement_returns_data():
from pandas import DataFrame
fs = make_fs()
data = DataFrame({"label": ["Revenue"], "value": [1]})
meta = DataFrame({"tag": ["x"]})
with (
patch.object(fs, "get_main_document_content", return_value="<html>x</html>"),
patch(
"openbb_sec.utils.non_xbrl_parser.parse_non_xbrl_statement",
return_value=(data, meta),
),
):
out_data, out_meta = fs._parse_non_xbrl_statement("income")
assert out_data is data and out_meta is meta
def test_download_statement_non_xbrl_delegates():
fs = make_fs()
with patch.object(fs, "_parse_non_xbrl_statement", return_value=("d", "m")) as p:
assert fs._download_statement("income") == ("d", "m")
p.assert_called_once_with("income")
def test_initialize_non_xbrl_already_initialized():
fs = make_fs(_non_xbrl_initialized=True)
fs._initialize_non_xbrl()
assert fs._non_xbrl_statements == {}
def test_initialize_non_xbrl_no_url_returns():
fs = make_fs()
with (
patch.object(fs, "get_embedded_document", return_value=None),
patch.object(fs, "get_main_document_url", return_value=None),
patch.object(fs, "get_main_document_content", return_value=None),
):
fs._initialize_non_xbrl()
assert fs._non_xbrl_initialized is False
def test_initialize_non_xbrl_inline_content():
from pandas import DataFrame
fs = make_fs()
with (
patch.object(fs, "get_embedded_document", return_value=None),
patch.object(fs, "get_main_document_url", return_value=None),
patch.object(
fs, "get_main_document_content", return_value="<html>inline</html>"
),
patch(
"openbb_sec.utils.non_xbrl_parser.find_all_statements",
return_value={"income": (DataFrame({"a": [1]}), DataFrame())},
),
patch(
"openbb_sec.utils.non_xbrl_parser.get_statement_names",
return_value={"income": "Income"},
),
patch("openbb_sec.utils.non_xbrl_parser.extract_toc", return_value={}),
patch(
"openbb_sec.utils.non_xbrl_parser.extract_text_blocks",
return_value={"note1": {"name": "Income Taxes", "text": "x"}},
),
patch("openbb_sec.utils.non_xbrl_parser.extract_items", return_value={}),
patch.object(fs, "_parse_exhibit_items"),
):
fs._initialize_non_xbrl()
assert fs._non_xbrl_initialized is True
assert fs._disclosures == {"note1": {"name": "Income Taxes", "text": "x"}}
def test_initialize_non_xbrl_from_embedded_main_doc():
from pandas import DataFrame
fs = make_fs()
data = DataFrame({"a": [1]})
meta = DataFrame({"tag": ["t"]})
with (
patch.object(
fs,
"get_embedded_document",
side_effect=lambda k: "<html>10k</html>" if k == "10-K" else None,
),
patch(
"openbb_sec.utils.non_xbrl_parser.find_all_statements",
return_value={"income": (data, meta)},
),
patch(
"openbb_sec.utils.non_xbrl_parser.get_statement_names",
return_value={"income": "Income"},
),
patch(
"openbb_sec.utils.non_xbrl_parser.extract_toc",
return_value={"1": "Business"},
),
patch(
"openbb_sec.utils.non_xbrl_parser.extract_text_blocks",
return_value={"note1": {"text": "x"}},
),
patch(
"openbb_sec.utils.non_xbrl_parser.extract_items",
return_value={"item_1": {"name": "Biz"}},
),
):
fs._initialize_non_xbrl()
assert fs._non_xbrl_statements["income"]["name"] == "Income"
assert fs._toc == {"1": "Business"}
assert fs._disclosures == {"note1": {"text": "x"}}
assert fs._items == {"item_1": {"name": "Biz"}}
assert fs._non_xbrl_initialized is True
def test_initialize_non_xbrl_downloads_main_doc_bytes():
from pandas import DataFrame
fs = make_fs()
with (
patch.object(fs, "get_embedded_document", return_value=None),
patch.object(
fs, "get_main_document_url", return_value="https://sec.gov/main.htm"
),
patch.object(fs, "_get_document", return_value=b"<html>doc</html>"),
patch(
"openbb_sec.utils.non_xbrl_parser.find_all_statements",
return_value={"balance": (DataFrame({"a": [1]}), DataFrame())},
),
patch("openbb_sec.utils.non_xbrl_parser.get_statement_names", return_value={}),
patch("openbb_sec.utils.non_xbrl_parser.extract_toc", return_value={}),
patch("openbb_sec.utils.non_xbrl_parser.extract_text_blocks", return_value={}),
patch("openbb_sec.utils.non_xbrl_parser.extract_items", return_value={}),
patch.object(fs, "_parse_exhibit_disclosures"),
patch.object(fs, "_parse_exhibit_items"),
):
fs._initialize_non_xbrl()
assert fs._non_xbrl_statements["balance"]["name"] == "Balance"
def test_initialize_non_xbrl_falls_back_to_exhibits():
fs = make_fs()
with (
patch.object(fs, "get_embedded_document", return_value="<html>main</html>"),
patch("openbb_sec.utils.non_xbrl_parser.find_all_statements", return_value={}),
patch("openbb_sec.utils.non_xbrl_parser.get_statement_names", return_value={}),
patch("openbb_sec.utils.non_xbrl_parser.extract_toc", return_value={}),
patch("openbb_sec.utils.non_xbrl_parser.extract_text_blocks", return_value={}),
patch("openbb_sec.utils.non_xbrl_parser.extract_items", return_value={}),
patch.object(fs, "_parse_exhibit_statements") as pes,
patch.object(fs, "_parse_exhibit_disclosures"),
patch.object(fs, "_parse_exhibit_items"),
):
fs._initialize_non_xbrl()
pes.assert_called_once()
def test_initialize_non_xbrl_warns_on_exception():
fs = make_fs()
with (
patch.object(fs, "get_embedded_document", side_effect=RuntimeError("boom")),
pytest.warns(UserWarning, match="Failed to parse non-XBRL"),
):
fs._initialize_non_xbrl()
def test_parse_exhibit_statements_from_embedded():
from pandas import DataFrame
fs = make_fs()
calls = {"find": lambda html: {"income": (DataFrame({"a": [1]}), DataFrame())}}
with patch.object(
fs,
"get_embedded_document",
side_effect=lambda k: "<html>x</html>" if k == "EX-13" else None,
):
fs._parse_exhibit_statements(calls["find"], {"income": "Income"})
assert fs._non_xbrl_statements["income"]["name"] == "Income"
def test_parse_exhibit_statements_download_priority():
from pandas import DataFrame
fs = make_fs(_document_urls=[{"type": "EX-13", "url": "https://sec.gov/ex13.htm"}])
with (
patch.object(fs, "get_embedded_document", return_value=None),
patch.object(fs, "_get_document", return_value=b"<html>ex</html>"),
):
fs._parse_exhibit_statements(
lambda html: {"income": (DataFrame({"a": [1]}), DataFrame())},
{"income": "Income"},
)
assert fs._non_xbrl_statements["income"]["name"] == "Income"
def test_parse_exhibit_statements_all_html_exhibits():
from pandas import DataFrame
fs = make_fs(_document_urls=[{"type": "EX-99", "url": "https://sec.gov/ex99.htm"}])
with (
patch.object(fs, "get_embedded_document", return_value=None),
patch.object(fs, "_get_document", return_value=b"<html>ex</html>"),
):
fs._parse_exhibit_statements(
lambda html: {"income": (DataFrame({"a": [1]}), DataFrame())},
{},
)
assert fs._non_xbrl_statements["income"]["name"] == "Income"
def test_parse_exhibit_disclosures_download_priority():
fs = make_fs(_document_urls=[{"type": "EX-13", "url": "https://sec.gov/ex13.htm"}])
with (
patch.object(fs, "get_embedded_document", return_value=None),
patch.object(fs, "_get_document", return_value=b"<html>ex</html>"),
):
fs._parse_exhibit_disclosures(lambda html: {"note": {"text": "t"}})
assert fs._disclosures == {"note": {"text": "t"}}
def test_parse_exhibit_items_download_priority():
fs = make_fs(_document_urls=[{"type": "EX-13", "url": "https://sec.gov/ex13.htm"}])
with (
patch.object(fs, "get_embedded_document", return_value=None),
patch.object(fs, "_get_document", return_value=b"<html>ex</html>"),
):
fs._parse_exhibit_items(
lambda html: {"item_1": {"name": "Biz"}},
lambda html: {"note": {"text": "t"}},
)
assert fs._items == {"item_1": {"name": "Biz"}}
assert fs._disclosures == {"note": {"text": "t"}}
def test_parse_exhibit_disclosures_from_embedded():
fs = make_fs()
with patch.object(
fs,
"get_embedded_document",
side_effect=lambda k: "<html>x</html>" if k == "EX-13" else None,
):
fs._parse_exhibit_disclosures(lambda html: {"note": {"text": "t"}})
assert fs._disclosures == {"note": {"text": "t"}}
def test_parse_exhibit_statements_embedded_raises_continues():
fs = make_fs()
with patch.object(fs, "get_embedded_document", return_value="<html>x</html>"):
fs._parse_exhibit_statements(
lambda html: (_ for _ in ()).throw(RuntimeError("boom")), {}
)
assert fs._non_xbrl_statements == {}
def test_parse_exhibit_disclosures_embedded_raises_continues():
fs = make_fs()
with patch.object(fs, "get_embedded_document", return_value="<html>x</html>"):
fs._parse_exhibit_disclosures(
lambda html: (_ for _ in ()).throw(RuntimeError("boom"))
)
assert fs._disclosures == {}
def test_parse_exhibit_items_embedded_raises_continues():
fs = make_fs()
with patch.object(fs, "get_embedded_document", return_value="<html>x</html>"):
fs._parse_exhibit_items(
lambda html: (_ for _ in ()).throw(RuntimeError("boom")),
lambda html: {},
)
assert fs._items == {}
def test_parse_exhibit_statements_download_raises_and_empty_url():
fs = make_fs(
_document_urls=[
{"type": "EX-13", "url": ""},
{"type": "EX-13.1", "url": "https://sec.gov/x.htm"},
]
)
with (
patch.object(fs, "get_embedded_document", return_value=None),
patch.object(fs, "_get_document", side_effect=RuntimeError("boom")),
):
fs._parse_exhibit_statements(lambda html: {}, {})
assert fs._non_xbrl_statements == {}
def test_parse_exhibit_statements_all_html_download_raises():
fs = make_fs(_document_urls=[{"type": "EX-99", "url": "https://sec.gov/x.htm"}])
with (
patch.object(fs, "get_embedded_document", return_value=None),
patch.object(fs, "_get_document", side_effect=RuntimeError("boom")),
):
fs._parse_exhibit_statements(lambda html: {}, {})
assert fs._non_xbrl_statements == {}
def test_parse_exhibit_statements_all_html_skips_non_html():
fs = make_fs(_document_urls=[{"type": "EX-99", "url": "https://sec.gov/x.pdf"}])
with (
patch.object(fs, "get_embedded_document", return_value=None),
patch.object(fs, "_get_document") as gd,
):
fs._parse_exhibit_statements(lambda html: {}, {})
gd.assert_not_called()
def test_parse_exhibit_disclosures_download_raises_and_empty_url():
fs = make_fs(
_document_urls=[
{"type": "EX-13", "url": ""},
{"type": "EX-13.1", "url": "https://sec.gov/x.htm"},
]
)
with (
patch.object(fs, "get_embedded_document", return_value=None),
patch.object(fs, "_get_document", side_effect=RuntimeError("boom")),
):
fs._parse_exhibit_disclosures(lambda html: {})
assert fs._disclosures == {}
def test_parse_exhibit_items_download_raises_and_empty_url():
fs = make_fs(
_document_urls=[
{"type": "EX-13", "url": ""},
{"type": "EX-13.1", "url": "https://sec.gov/x.htm"},
]
)
with (
patch.object(fs, "get_embedded_document", return_value=None),
patch.object(fs, "_get_document", side_effect=RuntimeError("boom")),
):
fs._parse_exhibit_items(lambda html: {}, lambda html: {})
assert fs._items == {}
def _metalinks_doc():
return {
"S1": {
"report": {
"": {},
"R1": {
"shortName": "Cover Page",
"longName": "Cover Page",
"groupType": "statement",
"menuCat": "Cover",
"uniqueAnchor": {"name": "dei:Doc", "unitRef": "U"},
},
"R2": {
"shortName": "Income",
"longName": "Income Statement",
"groupType": "statement",
"uniqueAnchor": {"name": "us-gaap:Revenue"},
},
"BadItem": None,
},
"tag": {
"us-gaap_Revenue": {
"xbrltype": "monetaryItemType",
"localname": "Revenue",
"crdr": "credit",
"calculation": {"http://x/role/Income": {"weight": "1"}},
"presentation": ["http://x/role/Income"],
"lang": {"en-US": {"role": {"label": "Revenue"}}},
"auth_ref": ["ref1", "missing"],
},
"EmptyTag": None,
},
}
}
def test_download_metalinks_no_url_returns():
fs = make_fs(_document_urls=[{"url": "https://sec.gov/x.htm"}])
fs._download_metalinks()
assert fs._resources == {}
def test_download_metalinks_already_loaded():
fs = make_fs(
_document_urls=[{"url": "https://sec.gov/MetaLinks.json"}],
_metalinks=_metalinks_doc(),
_std_ref={"ref1": {"Topic": "718", "SubTopic": "10"}},
)
fs._download_metalinks()
assert "us-gaap_Revenue" in fs._tags
assert fs._cover_page_url == "R1.htm"
assert fs._resources["r2"]["short_name"] == "Income"
def test_download_metalinks_downloads_when_empty():
fs = make_fs(_document_urls=[{"url": "https://sec.gov/MetaLinks.json"}])
doc = {"instance": _metalinks_doc(), "std_ref": {"ref1": {"Topic": "718"}}}
with patch.object(fs, "_get_document", return_value=doc):
fs._download_metalinks()
assert "us-gaap_Revenue" in fs._tags
def test_download_metalinks_skips_falsy_statement_item():
fs = make_fs(
_document_urls=[{"url": "https://sec.gov/MetaLinks.json"}],
_metalinks=_metalinks_doc(),
_resources={"empty": {}},
)
fs._download_metalinks()
assert fs._resources["empty"] == {}
def test_download_metalinks_raises_on_bad_structure():
fs = make_fs(
_document_urls=[{"url": "https://sec.gov/MetaLinks.json"}],
_metalinks={"S1": {"no_report_key": {}}},
)
with pytest.raises(RuntimeError, match="Failed to parse MetaLinks.json"):
fs._download_metalinks()
def _instance_doc_urls():
return [{"type": "EX-101.INS", "url": "https://sec.gov/x_htm.xml"}]
def test_download_xbrl_instance_empty_returns():
fs = make_fs(_document_urls=_instance_doc_urls())
parser = SimpleNamespace(parse_instance=lambda *a, **k: ({}, {}, {}))
with (
patch.object(fs, "_get_document", return_value=b"x"),
patch.object(fs, "_ensure_bytes", return_value=b"x"),
patch.object(mod, "XBRLParser", lambda: parser),
):
fs._download_xbrl_instance()
assert fs._instance == {}
def test_download_xbrl_instance_supplements_tags_and_textblock():
fs = make_fs(
_document_urls=_instance_doc_urls(),
_tags={"us-gaap_Revenue": {}},
)
contexts = {"ctx1": {"start": "2024-01-01"}}
units = {"U": "iso4217:USD"}
facts = {
"us-gaap_Revenue": [
{"value": "100", "label": "Revenue Label", "documentation": "Doc text"}
],
"us-gaap_InventoryTextBlock": [
{"value": "<p>note</p>", "label": "Inventory", "documentation": "d"}
],
"us-gaap_EmptyTag": [],
}
parser = SimpleNamespace(parse_instance=lambda *a, **k: (contexts, units, facts))
with (
patch.object(fs, "_get_document", return_value=b"x"),
patch.object(fs, "_ensure_bytes", return_value=b"x"),
patch.object(mod, "XBRLParser", lambda: parser),
patch.object(fs, "_clean_html_to_text", return_value="cleaned note"),
):
fs._download_xbrl_instance()
assert fs._period_context == contexts
assert fs._xbrl_units == units
assert fs._tags["us-gaap_Revenue"]["label"] == "Revenue Label"
assert fs._tags["us-gaap_Revenue"]["documentation"] == "Doc text"
assert fs._text_blocks["us-gaap_InventoryTextBlock"]["value"] == "cleaned note"
def test_download_xbrl_instance_no_instance_url():
fs = make_fs(_document_urls=[{"type": "GRAPHIC", "url": "x.jpg"}])
fs._download_xbrl_instance()
assert fs._instance == {}
def _schema_doc_urls():
return [
{"type": "EX-101.SCH", "url": "https://sec.gov/x.xsd"},
{"type": "EX-101.LAB", "url": "https://sec.gov/x_lab.xml"},
{"type": "EX-101.PRE", "url": "https://sec.gov/x_pre.xml"},
{"type": "EX-101.CAL", "url": "https://sec.gov/x_cal.xml"},
]
def test_build_schema_from_xml_fresh_full():
from openbb_sec.utils.xbrl_taxonomy_helper import XBRLNode
fs = make_fs(_document_urls=_schema_doc_urls())
elements = {"us-gaap_Revenue": {"xbrl_type": "monetary"}}
roles = [
{"document_number": "1"},
{"document_number": "1"},
{"document_number": ""},
]
embedded = ET.Element("linkbase")
node = XBRLNode(
element_id="us-gaap_Revenue",
label="Revenue",
order=1.0,
level=0,
parent_id=None,
preferred_label="http://x/role/totalLabel",
children=[
XBRLNode(
element_id="us-gaap_Child",
label="Child",
order=2.0,
level=1,
parent_id="us-gaap_Revenue",
)
],
)
class _Parser:
def parse_schema(self, *a, **k):
return elements, roles, embedded, [{"schemaLocation": "x"}]
def parse_label_linkbase(self, *a, **k):
return {"us-gaap_Revenue": {"label": "Revenue Label"}}
def parse_presentation(self, *a, **k):
return [node]
def parse_calculation(self, *a, **k):
return {
"us-gaap_Revenue": {"order": "1", "weight": "1", "parent_tag": "p"},
"us-gaap_New": {"weight": "2"},
}
with (
patch.object(fs, "_get_document", return_value=b"x"),
patch.object(fs, "_ensure_bytes", return_value=b"x"),
patch.object(mod, "XBRLParser", _Parser),
patch.object(fs, "_fetch_external_taxonomy_labels"),
):
fs._build_schema_from_xml()
assert fs._xsd["elements"] == elements
assert "r1" in fs._resources
assert fs._tags["us-gaap_Revenue"]["preferred_label"] == "totalLabel"
assert fs._tags["us-gaap_Revenue"]["weight"] == "1"
assert fs._tags["us-gaap_New"]["weight"] == "2"
assert fs._calcs
def test_build_schema_from_xml_cached_xsd():
fs = make_fs(
_document_urls=[{"type": "EX-101.SCH", "url": "https://sec.gov/x.xsd"}],
_xsd={
"elements": {
"us-gaap_Cash": {"xbrl_type": "monetary"},
"us-gaap_New": {"xbrl_type": "shares"},
},
"roles": [{"document_number": "1"}],
"imports": [],
},
_tags={"us-gaap_Cash": {"existing": True}},
)
with patch.object(fs, "_get_document") as gd:
fs._build_schema_from_xml()
gd.assert_not_called()
assert "r1" in fs._resources
assert fs._tags["us-gaap_Cash"]["existing"] is True
assert fs._tags["us-gaap_New"]["xbrl_type"] == "shares"
def test_build_schema_from_xml_exception_branches():
fs = make_fs(_document_urls=_schema_doc_urls())
class _BoomParser:
def parse_schema(self, *a, **k):
return {}, [], None, []
def parse_label_linkbase(self, *a, **k):
raise RuntimeError("lab boom")
def parse_presentation(self, *a, **k):
raise RuntimeError("pre boom")
def parse_calculation(self, *a, **k):
raise RuntimeError("cal boom")
with (
patch.object(fs, "_get_document", return_value=b"x"),
patch.object(fs, "_ensure_bytes", return_value=b"x"),
patch.object(mod, "XBRLParser", _BoomParser),
patch.object(fs, "_fetch_external_taxonomy_labels"),
):
fs._build_schema_from_xml()
assert fs._presentation == []
assert fs._calcs == {}
def test_build_schema_from_xml_embedded_linkbase_exception():
fs = make_fs(
_document_urls=[{"type": "EX-101.SCH", "url": "https://sec.gov/x.xsd"}]
)
call_count = {"n": 0}
class _Parser:
def parse_schema(self, *a, **k):
return {}, [], ET.Element("linkbase"), []
def parse_label_linkbase(self, *a, **k):
call_count["n"] += 1
raise RuntimeError("embedded boom")
with (
patch.object(fs, "_get_document", return_value=b"x"),
patch.object(fs, "_ensure_bytes", return_value=b"x"),
patch.object(mod, "XBRLParser", _Parser),
patch.object(fs, "_fetch_external_taxonomy_labels"),
):
fs._build_schema_from_xml()
assert call_count["n"] == 1
def test_parse_exhibit_items_from_embedded():
fs = make_fs()
with patch.object(
fs,
"get_embedded_document",
side_effect=lambda k: "<html>x</html>" if k == "EX-13" else None,
):
fs._parse_exhibit_items(
lambda html: {"item_1": {"name": "Biz"}},
lambda html: {"note": {"text": "t"}},
)
assert fs._items == {"item_1": {"name": "Biz"}}
assert fs._disclosures == {"note": {"text": "t"}}
def test_get_disclosures_info_skips_non_dict_resource():
"""Non-dict resource values are skipped in _get_disclosures_info."""
fs = make_fs(_resources={"r1": "not-a-dict"}, _items={"item_1": {"name": "Biz"}})
assert fs._get_disclosures_info() == {"item_1": {"name": "Biz"}}
def test_enrich_std_ref_full_reference():
"""_enrich_std_ref builds topic/subtopic/section names and an ASC reference."""
out = FinancialStatements._enrich_std_ref(
{
"Topic": "718",
"SubTopic": "10",
"Section": "50",
"Paragraph": "2",
}
)
assert out["topic_name"] == "Compensation—Stock Compensation"
assert out["subtopic_name"] == "Overall"
assert out["section_name"] == "Disclosure"
assert out["reference"].startswith("ASC 718-10-50-2")
assert "Compensation—Stock Compensation" in out["reference"]
def test_enrich_std_ref_no_topic_no_ref():
"""Without a topic, no reference label is produced."""
out = FinancialStatements._enrich_std_ref({})
assert "reference" not in out
assert "topic_name" not in out
def test_fetch_external_taxonomy_labels_merges_and_skips():
"""External taxonomy labels are merged, skipping missing urls and dup roles."""
fs = make_fs(_labels={"us-gaap_Cash": {"label": "Existing"}})
class _Parser:
def parse_label_linkbase(self, *a, **k):
return {
"us-gaap_Cash": {"label": "New", "documentation": "Doc"},
"us-gaap_New": {"label": "Brand"},
}
imports = [
{"schemaLocation": "no-label-url"},
{"schemaLocation": "https://fasb.org/us-gaap.xsd"},
]
with (
patch.object(
mod,
"get_label_url_for_import",
side_effect=lambda loc: (
"" if loc == "no-label-url" else "https://fasb.org/lab.xml"
),
),
patch.object(fs, "_get_document", return_value=b"x"),
patch.object(fs, "_ensure_bytes", return_value=b"x"),
patch.object(mod, "XBRLParser", _Parser),
):
fs._fetch_external_taxonomy_labels(imports)
assert fs._labels["us-gaap_Cash"]["label"] == "Existing"
assert fs._labels["us-gaap_Cash"]["documentation"] == "Doc"
assert fs._labels["us-gaap_New"] == {"label": "Brand"}
def test_fetch_external_taxonomy_labels_caches_on_miss():
"""A freshly parsed label linkbase is written to the cache when enabled."""
fs = make_fs(_labels={}, _use_cache=True)
saved: dict = {}
class _Parser:
def parse_label_linkbase(self, *a, **k):
return {"us-gaap_Cash": {"label": "Cash"}}
with (
patch.object(
mod, "get_label_url_for_import", return_value="https://fasb.org/lab.xml"
),
patch.object(fs, "_get_document", return_value=b"x"),
patch.object(fs, "_ensure_bytes", return_value=b"x"),
patch.object(mod, "XBRLParser", _Parser),
patch("openbb_sec.utils.cache._cache_get", return_value=None),
patch(
"openbb_sec.utils.cache._cache_set",
side_effect=lambda k, v, t: saved.update({k: v}),
),
):
fs._fetch_external_taxonomy_labels([{"schemaLocation": "fasb"}])
assert saved["xbrl-labels:https://fasb.org/lab.xml"] == {
"us-gaap_Cash": {"label": "Cash"}
}
assert fs._labels["us-gaap_Cash"] == {"label": "Cash"}
def test_fetch_external_taxonomy_labels_swallows_exception():
"""A parser error during external label fetch is swallowed."""
fs = make_fs(_labels={})
class _BoomParser:
def parse_label_linkbase(self, *a, **k):
raise RuntimeError("boom")
with (
patch.object(
mod, "get_label_url_for_import", return_value="https://sec.gov/lab.xml"
),
patch.object(fs, "_get_document", return_value=b"x"),
patch.object(fs, "_ensure_bytes", return_value=b"x"),
patch.object(mod, "XBRLParser", _BoomParser),
):
fs._fetch_external_taxonomy_labels([{"schemaLocation": "sec"}])
assert fs._labels == {}
def test_build_schema_from_xml_fresh_updates_existing_tag():
"""A schema element already present in tags is updated, not overwritten."""
fs = make_fs(
_document_urls=[{"type": "EX-101.SCH", "url": "https://sec.gov/x.xsd"}],
_tags={"us-gaap_Revenue": {"keep": True}},
)
elements = {"us-gaap_Revenue": {"xbrl_type": "monetary"}}
class _Parser:
def parse_schema(self, *a, **k):
return elements, [{"document_number": "1"}], None, []
with (
patch.object(fs, "_get_document", return_value=b"x"),
patch.object(fs, "_ensure_bytes", return_value=b"x"),
patch.object(mod, "XBRLParser", _Parser),
patch.object(fs, "_fetch_external_taxonomy_labels"),
):
fs._build_schema_from_xml()
assert fs._tags["us-gaap_Revenue"]["keep"] is True
assert fs._tags["us-gaap_Revenue"]["xbrl_type"] == "monetary"
def test_build_schema_from_xml_label_linkbase_merge():
"""The label linkbase merges new roles into _labels and onto tags."""
fs = make_fs(
_document_urls=[{"type": "EX-101.LAB", "url": "https://sec.gov/x_lab.xml"}],
_tags={"us-gaap_Revenue": {}},
_labels={"us-gaap_Revenue": {"label": "Old"}},
)
class _Parser:
def parse_label_linkbase(self, *a, **k):
return {
"us-gaap_Revenue": {"label": "New", "terseLabel": "Terse"},
"us-gaap_Other": {"label": "Other"},
}
with (
patch.object(fs, "_get_document", return_value=b"x"),
patch.object(fs, "_ensure_bytes", return_value=b"x"),
patch.object(mod, "XBRLParser", _Parser),
):
fs._build_schema_from_xml()
assert fs._labels["us-gaap_Revenue"]["label"] == "Old"
assert fs._labels["us-gaap_Revenue"]["terseLabel"] == "Terse"
assert fs._tags["us-gaap_Revenue"]["terseLabel"] == "Terse"
def test_download_xbrl_instance_textblock_uses_supplemented_tag():
"""A text-block fact gets a cleaned value attached to its supplemented tag."""
fs = make_fs(
_document_urls=_instance_doc_urls(),
_tags={},
)
facts = {
"us-gaap_InventoryTextBlock": [
{"value": "<p>note</p>", "label": "Inventory", "documentation": "Doc"}
],
}
parser = SimpleNamespace(parse_instance=lambda *a, **k: ({}, {}, facts))
with (
patch.object(fs, "_get_document", return_value=b"x"),
patch.object(fs, "_ensure_bytes", return_value=b"x"),
patch.object(mod, "XBRLParser", lambda: parser),
patch.object(fs, "_clean_html_to_text", return_value="clean"),
):
fs._download_xbrl_instance()
block = fs._text_blocks["us-gaap_InventoryTextBlock"]
assert block["value"] == "clean"
assert block["label"] == "Inventory"
assert block["documentation"] == "Doc"
def _download_statement_from_url_table(self, url, is_equity=False):
"""Return a crafted (table, meta) tuple for _download_statement tests."""
from pandas import DataFrame
table = DataFrame(
{
"Statement of Operations $ in Millions": [
"Net sales and revenues",
"Net income",
"Product revenue",
],
"2024-08-31 -- 12 Months Ended": [100, 50, 75],
"2023-08-31 -- 12 Months Ended": [90, 40, 60],
}
)
meta = DataFrame(
{
"Name": ["us-gaap_Revenues", "us-gaap_NetIncomeLoss"],
"Namespace Prefix": ["us-gaap", "us-gaap"],
"Data Type": ["xbrli:monetaryItemType", "xbrli:monetaryItemType"],
"Balance Type": ["credit", "credit"],
"Period Type": ["duration", "duration"],
"unit": ["iso4217:USD", "iso4217:USD"],
"decimals": ["-6", "-6"],
"weight": ["1", "1"],
"preferred_label": ["totalLabel", "label"],
"parent_tag": ["us-gaap_IncomeStatement", None],
}
)
return table, meta
def _full_income_fs():
"""FinancialStatements wired to exercise the full XBRL numeric pipeline."""
fs = make_fs(
_document_type="10-K",
_period_ending="2024-08-31",
_instance={
"us-gaap_Revenues": {
"context": [
{
"value": "100",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "iso4217:USD",
"decimals": "-6",
},
{
"value": "90",
"context_ref": "duration_2022_09_01_to_2023_08_31",
"unit": "iso4217:USD",
"decimals": "-6",
},
]
},
"us-gaap_NetIncomeLoss": {
"context": [
{
"value": "50",
"context_ref": "as_of_2024_08_31",
"unit": "iso4217:USD",
"decimals": "-6",
},
{
"value": "40",
"context_ref": "as_of_2023_08_31",
"unit": "iso4217:USD",
"decimals": "-6",
},
]
},
"us-gaap_RevenueFromProduct": {
"context": [
{
"value": "75",
"context_ref": "duration_2023_09_01_to_2024_08_31_us-gaap_ProductMember",
"unit": "iso4217:USD",
"decimals": "-6",
},
{
"value": "60",
"context_ref": "duration_2022_09_01_to_2023_08_31_us-gaap_ProductMember",
"unit": "iso4217:USD",
"decimals": "-6",
},
]
},
},
_tags={
"us-gaap_Revenues": {
"label": "Net sales and revenues",
"terseLabel": "Net sales and revenues",
"crdr": "credit",
"weight": "1",
"parent_tag": "us-gaap_IncomeStatement",
"preferred_label": "totalLabel",
"name": "Revenues",
"period_type": "duration",
},
"us-gaap_NetIncomeLoss": {
"label": "Net income",
"crdr": "credit",
"name": "NetIncomeLoss",
"period_type": "duration",
},
"us-gaap_RevenueFromProduct": {
"label": "Product revenue",
"crdr": "credit",
"name": "RevenueFromProduct",
},
"us-gaap_ProductMember": {"terseLabel": "Product line"},
},
_period_context={
"duration_2023_09_01_to_2024_08_31": {"start": "2023-09-01"},
},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
return fs
def test_download_statement_full_numeric_pipeline():
"""Drive _download_statement through label/unit/period/dimension stages."""
fs = _full_income_fs()
with patch.object(
fs,
"_download_statement_from_url",
new=_download_statement_from_url_table.__get__(fs),
):
data, meta = fs._download_statement("income")
cols = set(data.columns)
assert {
"order",
"tag",
"parent_tag",
"preferred_label",
"balance",
"weight",
"decimals",
"context_ref",
"period_beginning",
"period_ending",
"unit",
"label",
"value",
} <= cols
assert "us-gaap_Revenues" in set(data.tag.dropna())
assert not meta.empty
def test_download_statement_missing_tags_enrich_meta():
"""Tags present in the statement but absent from meta are added to output_meta."""
from pandas import DataFrame
fs = _full_income_fs()
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": ["Net sales and revenues"],
"2024-08-31 -- 12 Months Ended": [100],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, meta = fs._download_statement("income")
assert "us-gaap_Revenues" in set(meta.tag.dropna())
def test_download_statement_unit_and_decimals_from_instance():
"""Unit/decimals fall back to instance values when meta lacks the columns."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-K",
_period_ending="2024-08-31",
_instance={
"us-gaap_Assets": {
"context": [
{
"value": "200",
"context_ref": "as_of_2024_08_31",
"unit": "unit_Standard_iso4217_USD",
"decimals": "-6",
}
]
}
},
_tags={"us-gaap_Assets": {"label": "Total assets", "crdr": "debit"}},
_period_context={},
_resources={
"r1": {
"group": "statement",
"short_name": "Balance Sheet",
"url": "https://sec.gov/R2.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{"Balance Sheet $ in Millions": ["Total assets"], "2024-08-31": [200]}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("balance")
row = data[data.tag == "us-gaap_Assets"].iloc[0]
assert row.unit == "iso4217"
assert row.decimals == "-6"
def test_download_statement_period_beginning_from_duration_member():
"""Duration context refs yield a period_beginning derived from the ref."""
fs = _full_income_fs()
def _from_url(url, is_equity=False):
from pandas import DataFrame
table = DataFrame(
{
"Statement of Operations $ in Millions": ["Net sales and revenues"],
"2024-08-31 -- 12 Months Ended": [100],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
revenues = data[data.tag == "us-gaap_Revenues"]
assert (revenues.period_beginning == "2023-09-01").any()
def test_download_statement_dimension_relabel():
"""A generic revenue label resolving to a member context gets the member label."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-K",
_period_ending="2024-08-31",
_instance={
"us-gaap_Revenues": {
"context": [
{
"value": "75",
"context_ref": "duration_2023_09_01_to_2024_08_31_us-gaap_ProductMember",
"unit": "iso4217:USD",
"decimals": "-6",
}
]
},
},
_tags={
"us-gaap_Revenues": {
"label": "Net sales and revenues",
"crdr": "credit",
"name": "Revenues",
},
"us-gaap_ProductMember": {"terseLabel": "Product line"},
},
_period_context={},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": ["Net sales and revenues"],
"2024-08-31 -- 12 Months Ended": [75],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
assert "Product line" in set(data.label)
def test_download_statement_meta_column_branches():
"""Unit/decimals/weight/balance/preferred_label/parent_tag read from output_meta."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-K",
_period_ending="2024-08-31",
_instance={
"us-gaap_Revenues": {
"context": [
{
"value": "100",
"context_ref": "duration_2023_09_01_to_2024_08_31",
}
]
}
},
_tags={
"us-gaap_Revenues": {"label": "Net sales and revenues", "name": "Revenues"}
},
_period_context={},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": ["Net sales and revenues"],
"2024-08-31 -- 12 Months Ended": [100],
}
)
meta = DataFrame(
{
"Name": ["us-gaap_Revenues"],
"Namespace Prefix": ["us-gaap"],
"Data Type": ["xbrli:monetaryItemType"],
"Balance Type": ["credit"],
"Period Type": ["duration"],
"unit": ["iso4217:USD / shares"],
"decimals": ["-3"],
"weight": ["1"],
"preferred_label": ["totalLabel"],
"parent_tag": ["us-gaap_IncomeStatement"],
}
)
return table, meta
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
row = data[data.tag == "us-gaap_Revenues"].iloc[0]
assert row.unit == "USDPerShares"
assert row.decimals == "-3"
assert row.weight == "1"
assert row.balance == "credit"
assert row.preferred_label == "totalLabel"
assert row.parent_tag == "us-gaap_IncomeStatement"
def test_download_statement_normalize_instance_key_and_word_match():
"""Instance keys normalize by local name and apply_fix_tag word-matches labels."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-Q",
_period_ending="2024-05-31",
_instance={
"srt_Revenues": {
"context": [
{
"value": "100",
"context_ref": "duration_2024_03_01_to_2024_05_31",
"unit": "iso4217:USD",
"decimals": "-6",
}
]
},
"us-gaap_OperatingExpenses": {
"context": [
{
"value": "55",
"context_ref": "duration_2024_03_01_to_2024_05_31",
"unit": "iso4217:USD",
"decimals": "-6",
}
]
},
"us-gaap_OtherExpenses": {
"context": [
{
"value": "55",
"context_ref": "duration_2024_03_01_to_2024_05_31",
"unit": "iso4217:USD",
"decimals": "-6",
}
]
},
},
_tags={
"us-gaap_Revenues": {
"label": "Net sales and revenues",
"crdr": "credit",
"name": "Revenues",
},
"us-gaap_OperatingExpenses": {
"label": "Total operating expenses incurred",
"crdr": "debit",
"name": "OperatingExpenses",
},
"us-gaap_OtherExpenses": {
"label": "Unrelated other charges",
"crdr": "debit",
"name": "OtherExpenses",
},
},
_period_context={},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": [
"Net sales and revenues",
"Operating expenses incurred during period",
],
"2024-05-31 -- 3 Months Ended": [100, 55],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
tags = set(data.tag.dropna())
assert "us-gaap_Revenues" in tags
assert "us-gaap_OperatingExpenses" in tags
def test_download_statement_period_beginning_span_branch():
"""A row without a resolvable context derives period_beginning from the column span."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-Q",
_period_ending="2024-05-31",
_instance={
"us-gaap_Revenues": {
"context": [
{
"value": "100",
"context_ref": "duration_2024_03_01_to_2024_05_31",
"unit": "iso4217:USD",
}
]
},
"us-gaap_OperatingIncomeLoss": {
"context": [
{
"value": "55",
"context_ref": "c-3",
"unit": "iso4217:USD",
}
]
},
},
_tags={
"us-gaap_Revenues": {"label": "Net sales and revenues", "crdr": "credit"},
"us-gaap_OperatingIncomeLoss": {
"label": "Operating income",
"crdr": "credit",
},
},
_period_context={},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": [
"Net sales and revenues",
"Operating income",
],
"2024-05-31 -- 3 Months Ended": [100, 55],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
op_row = data[data.label == "Operating income"].iloc[0]
assert op_row.period_beginning == "2024-03-01"
def test_download_statement_period_beginning_10k_year_shift():
"""A 10-K row with no span and no context falls back to a one-year-prior beginning."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-K",
_period_ending="2024-08-31",
_instance={
"us-gaap_Revenues": {
"context": [
{
"value": "100",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "iso4217:USD",
}
]
},
"us-gaap_OperatingIncomeLoss": {
"context": [
{
"value": "55",
"context_ref": "c-3",
"unit": "iso4217:USD",
}
]
},
},
_tags={
"us-gaap_Revenues": {"label": "Net sales and revenues", "crdr": "credit"},
"us-gaap_OperatingIncomeLoss": {
"label": "Operating income",
"crdr": "credit",
},
},
_period_context={},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": [
"Net sales and revenues",
"Operating income",
],
"2024-08-31": [100, 55],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
op_row = data[data.label == "Operating income"].iloc[0]
assert op_row.period_beginning == "2023-09-01"
def test_download_statement_value_to_tags_fix():
"""A mislabeled row tag is corrected by numeric value matching."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-K",
_period_ending="2024-08-31",
_instance={
"us-gaap_Revenues": {
"context": [
{
"value": "100",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "iso4217:USD",
"decimals": "-6",
}
]
},
},
_tags={
"us-gaap_Revenues": {
"label": "Net sales and revenues",
"crdr": "credit",
"name": "Revenues",
},
"us-gaap_WrongLabelTag": {
"label": "Net sales and revenues",
"crdr": "credit",
"name": "WrongLabelTag",
},
},
_period_context={"duration_2023_09_01_to_2024_08_31": {"start": "2023-09-01"}},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": ["Net sales and revenues"],
"2024-08-31 -- 12 Months Ended": [100],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
assert "us-gaap_Revenues" in set(data.tag.dropna())
def test_download_statement_from_url_single_period():
"""_download_statement_from_url processes a single-period millions table."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-Q",
_instance={
"us-gaap_Revenues": {"context": [{"unit": "iso4217:USD", "decimals": "-6"}]}
},
_tags={"us-gaap_Revenues": {"weight": "1", "name": "Revenues"}},
)
raw = DataFrame(
{
0: ["Statement of Operations $ in Millions", "Net sales and revenues"],
1: ["3 Months Ended Aug. 31, 2024", "100"],
}
)
raw = raw.set_axis(
["Statement of Operations $ in Millions", "Aug. 31, 2024"], axis=1
)
table = DataFrame(
{
"Statement of Operations $ in Millions": ["Net sales and revenues"],
"Aug. 31, 2024": ["100"],
}
)
meta = DataFrame(
{
0: ["Name", "Namespace Prefix", "Balance Type"],
1: ["us-gaap_Revenues", "us-gaap", "credit"],
}
)
with patch.object(fs, "_get_document", return_value=[table, meta]):
out_table, out_meta = fs._download_statement_from_url("https://sec.gov/R4.htm")
assert "12 Months Ended" not in out_table.columns
assert out_table.iloc[0, 0] == "Net sales and revenues"
def test_download_statement_from_url_equity_branch():
"""_download_statement_from_url leaves columns intact for equity statements."""
from pandas import DataFrame
fs = make_fs(_document_type="10-K", _instance={}, _tags={})
table = DataFrame(
{
"Statement of Equity": ["Balance"],
"Common Stock": ["500"],
"Retained Earnings": ["300"],
}
)
with patch.object(fs, "_get_document", return_value=[table]):
out_table, out_meta = fs._download_statement_from_url(
"https://sec.gov/R7.htm", is_equity=True
)
assert "Common Stock" in out_table.columns
assert out_meta.empty
def test_download_statement_from_url_multiindex_two_periods():
"""A 5-column multiindex 10-K table is split into two period frames and merged."""
from pandas import DataFrame, MultiIndex
fs = make_fs(
_document_type="10-K",
_instance={
"us-gaap_Revenues": {"context": [{"unit": "iso4217:USD", "decimals": "-6"}]}
},
_tags={"us-gaap_Revenues": {"weight": "1", "name": "Revenues"}},
)
cols = MultiIndex.from_tuples(
[
("Income Statement $ in Millions", ""),
("12 Months Ended", "Aug. 31, 2024"),
("12 Months Ended", "Aug. 31, 2023"),
("12 Months Ended", "Aug. 31, 2022"),
("12 Months Ended", "Aug. 31, 2021"),
]
)
table = DataFrame(
[["Net sales and revenues", "100", "90", "80", "70"]], columns=cols
)
meta = DataFrame({0: ["Name", "Balance Type"], 1: ["us-gaap_Revenues", "credit"]})
with patch.object(fs, "_get_document", return_value=[table, meta]):
out_table, _ = fs._download_statement_from_url("https://sec.gov/R4.htm")
assert out_table.iloc[0, 0] == "Net sales and revenues"
def test_download_statement_from_url_three_col_multiindex():
"""A 3-column multiindex table is flattened with the period dropped to one level."""
from pandas import DataFrame, MultiIndex
fs = make_fs(
_document_type="10-Q",
_instance={
"us-gaap_Revenues": {"context": [{"unit": "iso4217:USD", "decimals": "-6"}]}
},
_tags={"us-gaap_Revenues": {"weight": "1", "name": "Revenues"}},
)
cols = MultiIndex.from_tuples(
[
("Income Statement $ in Millions", ""),
("3 Months Ended", "Aug. 31, 2024"),
("3 Months Ended", "Aug. 31, 2023"),
]
)
table = DataFrame([["Net sales and revenues", "100", "90"]], columns=cols)
meta = DataFrame({0: ["Name", "Balance Type"], 1: ["us-gaap_Revenues", "credit"]})
with patch.object(fs, "_get_document", return_value=[table, meta]):
out_table, _ = fs._download_statement_from_url("https://sec.gov/R4.htm")
assert out_table.iloc[0, 0] == "Net sales and revenues"
def test_download_statement_from_url_shares_and_per_share_rows():
"""Per-share and shares rows are formatted with their own multipliers."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-Q",
_instance={},
_tags={},
)
table = DataFrame(
{
"Income Statement $ in Millions, shares in Thousands": [
"Net sales and revenues",
"Earnings per share - basic (in dollars per share)",
"Weighted average shares outstanding",
"[1] footnote row",
],
"Aug. 31, 2024": ["100", "2.50", "40", "5"],
}
)
meta = DataFrame(
{
0: ["Name", "Balance Type"],
1: ["us-gaap_Revenues", "na"],
}
)
with patch.object(fs, "_get_document", return_value=[table, meta]):
out_table, _ = fs._download_statement_from_url("https://sec.gov/R4.htm")
labels = out_table.iloc[:, 0].tolist()
assert "[1] footnote row" not in labels
assert "Net sales and revenues" in labels
def test_download_statement_from_url_five_col_two_periods_merge():
"""A non-10-K 5-column multiindex splits into two periods that are merged."""
from pandas import DataFrame, MultiIndex
fs = make_fs(
_document_type="10-Q",
_instance={
"us-gaap_Revenues": {"context": [{"unit": "iso4217:USD", "decimals": "-6"}]}
},
_tags={"us-gaap_Revenues": {"weight": "1", "name": "Revenues"}},
)
cols = MultiIndex.from_tuples(
[
("Income Statement $ in Millions", ""),
("3 Months Ended", "Aug. 31, 2024"),
("3 Months Ended", "Aug. 31, 2023"),
("9 Months Ended", "Aug. 31, 2024"),
("9 Months Ended", "Aug. 31, 2023"),
]
)
table = DataFrame(
[["Net sales and revenues", "100", "90", "300", "280"]], columns=cols
)
meta = DataFrame({0: ["Name", "Balance Type"], 1: ["us-gaap_Revenues", "credit"]})
with patch.object(fs, "_get_document", return_value=[table, meta]):
out_table, _ = fs._download_statement_from_url("https://sec.gov/R4.htm")
assert out_table.iloc[0, 0] == "Net sales and revenues"
assert len(out_table.columns) == 5
def test_download_statement_from_url_no_period_and_bad_values():
"""A 3-col multiindex with an empty period level uses bare dates and drops bad rows."""
from pandas import DataFrame, MultiIndex
fs = make_fs(_document_type="10-Q", _instance={}, _tags={})
cols = MultiIndex.from_tuples(
[
("Income Statement $ in Millions", ""),
("", "Aug. 31, 2024"),
("", "Aug. 31, 2023"),
]
)
table = DataFrame(
[
["Net sales and revenues", "100", "90"],
["[bracket value row]", "[1]", "5"],
["Non-numeric row", "not-a-number", "10"],
],
columns=cols,
)
meta = DataFrame({0: ["Name", "Balance Type"], 1: ["us-gaap_Revenues", "credit"]})
with patch.object(fs, "_get_document", return_value=[table, meta]):
out_table, _ = fs._download_statement_from_url("https://sec.gov/R4.htm")
labels = out_table.iloc[:, 0].tolist()
assert " -- " not in str(out_table.columns[1])
assert "Net sales and revenues" in labels
bad = out_table[out_table.iloc[:, 0] == "Non-numeric row"]
assert bad.iloc[0, 1] == "--"
def test_download_statement_from_url_drops_duplicate_column():
"""A duplicated value column (name and name.1) is dropped before processing."""
from pandas import DataFrame
fs = make_fs(_document_type="10-Q", _instance={}, _tags={})
table = DataFrame(
[
["Income Statement $ in Millions", "Aug. 31, 2024", "Aug. 31, 2024"],
["Net sales and revenues", "100", "100"],
]
)
table.columns = ["Aug. 31, 2024", "Aug. 31, 2024", "Aug. 31, 2024.1"]
table = DataFrame(
{
"Aug. 31, 2024": ["Net sales and revenues"],
"Aug. 31, 2024.1": ["100"],
}
)
meta = DataFrame({0: ["Name", "Balance Type"], 1: ["us-gaap_Revenues", "credit"]})
with patch.object(fs, "_get_document", return_value=[table, meta]):
out_table, _ = fs._download_statement_from_url("https://sec.gov/R4.htm")
assert out_table.iloc[0, 0] == "Net sales and revenues"
def test_download_statement_from_url_drops_bracket_meta_columns():
"""Meta columns whose header starts with '[' are dropped from the item map."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-Q",
_instance={"us-gaap_Revenues": {"context": [{}]}},
_tags={"us-gaap_Revenues": {"name": "Revenues"}},
)
table = DataFrame(
{
"Income Statement $ in Millions": ["Net sales and revenues"],
"Aug. 31, 2024 Ended": ["100"],
}
)
meta = DataFrame(
{
0: ["Name", "Balance Type", "[Footnote]"],
1: ["us-gaap_Revenues", "credit", "[1]"],
}
)
with patch.object(fs, "_get_document", return_value=[table, meta]):
_, out_meta = fs._download_statement_from_url("https://sec.gov/R4.htm")
if not out_meta.empty:
assert "[Footnote]" not in out_meta.columns
def _fix_tag_fs(tags, instance):
"""FinancialStatements wired for apply_fix_tag cascade tests."""
return make_fs(
_document_type="10-K",
_period_ending="2024-08-31",
_instance=instance,
_tags=tags,
_period_context={},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
def _fix_tag_run(fs, label, value):
"""Drive _download_statement with a single-row table for a label/value."""
from pandas import DataFrame
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": [label],
"2024-08-31 -- 12 Months Ended": [value],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
return data[data.label == label].tag.tolist()
def test_apply_fix_tag_exact_label_cascade():
"""A wrong initial tag is corrected to the candidate with the exact label."""
fs = _fix_tag_fs(
tags={
"us-gaap_RevenueExact": {
"label": "Revenue",
"crdr": "credit",
"name": "RevenueExact",
},
"us-gaap_RevA": {
"label": "Receipts total",
"crdr": "credit",
"name": "RevA",
},
"us-gaap_RevB": {"label": "Revenue", "crdr": "credit", "name": "RevB"},
},
instance={
"us-gaap_RevenueExact": {
"context": [
{
"value": "90",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "iso4217:USD",
}
]
},
"us-gaap_RevA": {
"context": [
{
"value": "100",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "iso4217:USD",
}
]
},
"us-gaap_RevB": {
"context": [
{
"value": "100",
"context_ref": "as_of_2024_08_31",
"unit": "iso4217:USD",
}
]
},
},
)
assert _fix_tag_run(fs, "Revenue", 100) == ["us-gaap_RevB"]
def test_apply_fix_tag_word_overlap_cascade():
"""A wrong initial tag is corrected by a two-word label overlap."""
fs = _fix_tag_fs(
tags={
"us-gaap_Init": {
"label": "Cost of revenue",
"crdr": "debit",
"name": "Init",
},
"us-gaap_W1": {
"label": "Total cost of products sold",
"crdr": "debit",
"name": "W1",
},
"us-gaap_W2": {
"label": "Unrelated charges incurred",
"crdr": "debit",
"name": "W2",
},
},
instance={
"us-gaap_Init": {
"context": [
{
"value": "90",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "iso4217:USD",
}
]
},
"us-gaap_W1": {
"context": [
{
"value": "50",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "iso4217:USD",
}
]
},
"us-gaap_W2": {
"context": [
{
"value": "50",
"context_ref": "as_of_2024_08_31",
"unit": "iso4217:USD",
}
]
},
},
)
assert _fix_tag_run(fs, "Cost of products sold", 50) == ["us-gaap_W1"]
def test_apply_fix_tag_name_words_cascade():
"""A wrong initial tag is corrected by element-name word overlap."""
fs = _fix_tag_fs(
tags={
"us-gaap_Start": {
"label": "Inventory balance",
"crdr": "debit",
"name": "Start",
},
"us-gaap_InventoryNet": {
"label": "Goods on hand",
"crdr": "debit",
"name": "InventoryNet",
},
"us-gaap_OtherX": {
"label": "Misc holdings",
"crdr": "debit",
"name": "OtherX",
},
},
instance={
"us-gaap_Start": {
"context": [
{
"value": "90",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "iso4217:USD",
}
]
},
"us-gaap_InventoryNet": {
"context": [
{
"value": "70",
"context_ref": "as_of_2024_08_31",
"unit": "iso4217:USD",
}
]
},
"us-gaap_OtherX": {
"context": [
{
"value": "70",
"context_ref": "as_of_2024_08_31",
"unit": "iso4217:USD",
}
]
},
},
)
assert _fix_tag_run(fs, "Inventory", 70) == ["us-gaap_InventoryNet"]
def test_apply_fix_tag_no_match_keeps_initial():
"""When no cascade tier matches, the initial tag is kept unchanged."""
from pandas import DataFrame
fs = _fix_tag_fs(
tags={
"us-gaap_Start": {"label": "Net total", "crdr": "credit", "name": "Start"},
"srt_AlphaXxxx": {
"label": "Greek letters here",
"crdr": "credit",
"name": "AlphaXxxx",
},
"srt_BetaYyyy": {
"label": "Other letters here",
"crdr": "credit",
"name": "BetaYyyy",
},
"us-gaap_Cash": {
"label": "Cash and equivalents",
"crdr": "debit",
"name": "Cash",
},
},
instance={
"us-gaap_Start": {
"context": [
{
"value": "90",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "iso4217:USD",
}
]
},
"srt_AlphaXxxx": {
"context": [
{
"value": "33",
"context_ref": "as_of_2024_08_31",
"unit": "iso4217:USD",
}
]
},
"srt_BetaYyyy": {
"context": [
{
"value": "33",
"context_ref": "as_of_2024_08_31",
"unit": "iso4217:USD",
}
]
},
"us-gaap_Cash": {
"context": [
{
"value": "200",
"context_ref": "as_of_2024_08_31",
"unit": "iso4217:USD",
}
]
},
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": [
"Net total",
"Cash and equivalents",
],
"2024-08-31 -- 12 Months Ended": [33, 200],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
net_tag = data[data.label == "Net total"].tag.tolist()
assert net_tag[0] in (None, "us-gaap_Start")
def test_format_value_int_float_and_passthrough():
"""format_value coerces ints, keeps non-integer floats, and drops empties."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-K",
_period_ending="2024-08-31",
_instance={
"us-gaap_EarningsPerShare": {
"context": [
{
"value": "2.5",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "iso4217:USD / shares",
}
]
}
},
_tags={
"us-gaap_EarningsPerShare": {
"label": "Diluted earnings per share (in dollars per share)",
"crdr": "credit",
}
},
_period_context={},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": [
"Diluted earnings per share (in dollars per share)",
],
"2024-08-31 -- 12 Months Ended": [2.5],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
assert 2.5 in set(data.value.dropna())
def test_download_statement_normalized_label_match():
"""A 'Total assets' table label matches a tag labelled 'Assets, Total'."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-K",
_period_ending="2024-08-31",
_instance={
"us-gaap_Assets": {
"context": [
{
"value": "200",
"context_ref": "as_of_2024_08_31",
"unit": "iso4217:USD",
}
]
}
},
_tags={
"us-gaap_Assets": {
"label": "Assets, Total",
"crdr": "debit",
"name": "Assets",
}
},
_period_context={},
_resources={
"r1": {
"group": "statement",
"short_name": "Balance Sheet",
"url": "https://sec.gov/R2.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{"Balance Sheet $ in Millions": ["Total assets"], "2024-08-31": [200]}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("balance")
assert data[data.label == "Total assets"].tag.tolist() == ["us-gaap_Assets"]
def test_download_statement_check_unit_variants_from_meta():
"""check_unit handles divide, two-part, and passthrough unit strings from meta."""
from pandas import DataFrame
tags = {
"us-gaap_EpsA": {"label": "Diluted EPS", "name": "EpsA"},
"us-gaap_EpsB": {"label": "Basic EPS", "name": "EpsB"},
"us-gaap_PureC": {"label": "Tax rate", "name": "PureC"},
"us-gaap_PlainD": {"label": "Plain value", "name": "PlainD"},
}
instance = {
"us-gaap_EpsA": {
"context": [
{
"value": "2",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "x",
}
]
},
"us-gaap_EpsB": {
"context": [
{
"value": "3",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "x",
}
]
},
"us-gaap_PureC": {
"context": [
{
"value": "4",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "x",
}
]
},
"us-gaap_PlainD": {
"context": [
{
"value": "5",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "x",
}
]
},
}
fs = make_fs(
_document_type="10-K",
_period_ending="2024-08-31",
_instance=instance,
_tags=tags,
_period_context={},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
meta = DataFrame(
{
"Name": ["us-gaap_EpsA", "us-gaap_EpsB", "us-gaap_PureC", "us-gaap_PlainD"],
"Namespace Prefix": ["us-gaap"] * 4,
"Data Type": ["xbrli:perShareItemType"] * 4,
"Balance Type": ["na", "na", "na", "na"],
"Period Type": ["duration"] * 4,
"unit": ["unit_Divide_iso4217_Shares", "u_USD", "pure", "rawunit"],
}
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": [
"Diluted EPS",
"Basic EPS",
"Tax rate",
"Plain value",
],
"2024-08-31 -- 12 Months Ended": [2, 3, 4, 5],
}
)
return table, meta
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
units = dict(zip(data.tag, data.unit))
assert units["us-gaap_EpsA"] == "iso4217perShares"
assert units["us-gaap_EpsB"] == "USD"
assert units["us-gaap_PureC"] == "pure"
assert units["us-gaap_PlainD"] == "rawunit"
assert set(data.balance.dropna()) <= {None}
def test_download_statement_context_ref_and_orphan_edges():
"""apply_context_ref skips malformed entries and unmatched/empty-instance tags."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-K",
_period_ending="2024-08-31",
_instance={
"us-gaap_Mixed": {
"context": [
{},
{"context_ref": "", "value": "5"},
{"context_ref": "longref_aaaa", "value": "--"},
{"context_ref": "longref_bbbb", "value": "abc"},
{
"context_ref": "longref_good",
"value": "50",
"unit": "iso4217:USD",
},
]
},
"prefix_OrphanLocal": {
"context": [
{
"context_ref": "duration_2023_09_01_to_2024_08_31",
"value": "777",
"unit": "iso4217:USD",
}
]
},
},
_tags={
"us-gaap_Mixed": {"label": "Mixed line", "crdr": "credit", "name": "Mixed"},
"us-gaap_FooZ": {"label": "Foo line", "crdr": "credit", "name": "FooZ"},
},
_period_context={},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": [
"Mixed line",
"Foo line",
"Unmatched label here",
],
"2024-08-31 -- 12 Months Ended": [50, 60, "--"],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
refs = dict(zip(data.label, data.context_ref))
assert refs["Mixed line"] == "longref_good"
assert refs["Foo line"] is None
assert refs["Unmatched label here"] is None
def test_download_statement_context_ref_month_words_and_duration_match():
from pandas import DataFrame
fs = make_fs(
_document_type="10-Q",
_period_ending="2024-02-29",
_instance={
"us-gaap_WordMonths": {
"context": [
{
"context_ref": "ctx_bad",
"value": "10",
"start": "2023-11-01",
"end": "2024-02-29",
"unit": "iso4217:USD",
},
{
"context_ref": "ctx_good",
"value": "10",
"start": "2024-01-01",
"end": "2024-02-29",
"unit": "iso4217:USD",
},
]
}
},
_tags={
"us-gaap_WordMonths": {
"label": "Word months line",
"crdr": "credit",
"name": "WordMonths",
}
},
_period_context={
"ctx_bad": {"start": "2023-11-01"},
"ctx_good": {"start": "2024-01-01"},
},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": ["Word months line"],
"2024-02-29 -- Two Months Ended": [10],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
assert data.iloc[0].context_ref == "ctx_good"
def test_download_statement_period_ending_from_context_duration():
from pandas import DataFrame
fs = make_fs(
_document_type="10-Q",
_period_ending="2024-12-31",
_instance={
"us-gaap_DurationRow": {
"context": [
{
"context_ref": "ctx_duration",
"value": "100",
"unit": "iso4217:USD",
}
]
}
},
_tags={
"us-gaap_DurationRow": {
"label": "Duration row",
"crdr": "credit",
"name": "DurationRow",
}
},
_period_context={
"ctx_duration": {
"start": "2024-01-01",
"end": "2024-12-31",
"period_type": "duration",
}
},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": ["Duration row"],
"2024-12-31 -- 12 Months Ended": [100],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
assert data.iloc[0].period_ending == "2024-12-31 -- 12 Months Ended"
def test_download_statement_period_ending_from_context_instant():
from pandas import DataFrame
fs = make_fs(
_document_type="10-Q",
_period_ending="2024-12-31",
_instance={
"us-gaap_InstantRow": {
"context": [
{
"context_ref": "ctx_instant",
"value": "100",
"unit": "iso4217:USD",
}
]
}
},
_tags={
"us-gaap_InstantRow": {
"label": "Instant row",
"crdr": "credit",
"name": "InstantRow",
}
},
_period_context={
"ctx_instant": {
"start": None,
"end": "2024-12-31",
"period_type": "instant",
}
},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": ["Instant row"],
"2024-12-31 -- 12 Months Ended": [100],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
assert data.iloc[0].period_ending == "2024-12-31"
def test_download_statement_period_beginning_context_heuristics():
"""Plain context refs fall through to length heuristics or yield no beginning."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-Q",
_period_ending="2024-05-31",
_instance={
"us-gaap_Heur23": {
"context": [
{
"context_ref": "20240831_20230901_xxxxx",
"value": "11",
"unit": "iso4217:USD",
}
]
},
"us-gaap_Heur12": {
"context": [
{
"context_ref": "ab_20240831x",
"value": "12",
"unit": "iso4217:USD",
}
]
},
"us-gaap_NoneRow": {
"context": [
{
"context_ref": "plainrefnomatch",
"value": "13",
"unit": "iso4217:USD",
}
]
},
},
_tags={
"us-gaap_Heur23": {
"label": "Heur23 line",
"crdr": "credit",
"name": "Heur23",
},
"us-gaap_Heur12": {
"label": "Heur12 line",
"crdr": "credit",
"name": "Heur12",
},
"us-gaap_NoneRow": {
"label": "None row line",
"crdr": "credit",
"name": "NoneRow",
},
},
_period_context={},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": [
"Heur23 line",
"Heur12 line",
"None row line",
],
"2024-05-31": [11, 12, 13],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
begins = dict(zip(data.label, data.period_beginning))
assert begins["Heur23 line"] == "20230901"
assert begins["Heur12 line"] == "20240831x"
assert begins["None row line"] is None
def test_download_statement_dimension_member_without_terse_label():
"""A generic label whose member lacks a terse label keeps the original label."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-K",
_period_ending="2024-08-31",
_instance={
"us-gaap_Revenues": {
"context": [
{
"value": "75",
"context_ref": "duration_2023_09_01_to_2024_08_31_us-gaap_ProductMember",
"unit": "iso4217:USD",
}
]
}
},
_tags={
"us-gaap_Revenues": {"label": "Net sales and revenues", "crdr": "credit"},
"us-gaap_ProductMember": {"label": "Product line"},
},
_period_context={},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": ["Net sales and revenues"],
"2024-08-31 -- 12 Months Ended": [75],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
assert "Net sales and revenues" in set(data.label)
assert "Product line" not in set(data.label)
def test_download_statement_value_error_and_na_tag_unit():
"""A 'na'-keyed tag yields no unit and a malformed instance value is skipped."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-K",
_period_ending="2024-08-31",
_instance={
"us-gaap_Other": {
"context": [
{
"value": "1.2.3",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "iso4217:USD",
}
]
},
"na": {
"context": [
{
"value": "5",
"context_ref": "as_of_2024_08_31",
"unit": "iso4217:USD",
}
]
},
},
_tags={
"us-gaap_Other": {"label": "Other line", "crdr": "credit", "name": "Other"},
"na": {"label": "Na line", "crdr": "credit", "name": "Na"},
},
_period_context={},
_resources={
"r1": {
"group": "statement",
"short_name": "Statement of Operations",
"url": "https://sec.gov/R4.htm",
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": ["Na line"],
"2024-08-31 -- 12 Months Ended": [5],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
na_row = data[data.tag == "na"].iloc[0]
assert na_row.unit is None
def test_apply_fix_tag_single_tags_matching():
"""When only one candidate is a known tag, apply_fix_tag selects it last."""
from pandas import DataFrame
fs = _fix_tag_fs(
tags={
"us-gaap_Start": {"label": "Net total", "crdr": "credit", "name": "Start"},
"srt_AlphaXxxx": {
"label": "Greek letters here",
"crdr": "credit",
"name": "AlphaXxxx",
},
},
instance={
"us-gaap_Start": {
"context": [
{
"value": "90",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "iso4217:USD",
}
]
},
"srt_AlphaXxxx": {
"context": [
{
"value": "33",
"context_ref": "as_of_2024_08_31",
"unit": "iso4217:USD",
}
]
},
"srt_NotInTags": {
"context": [
{
"value": "33",
"context_ref": "as_of_2024_08_31",
"unit": "iso4217:USD",
}
]
},
"us-gaap_Cash": {
"context": [
{
"value": "200",
"context_ref": "as_of_2024_08_31",
"unit": "iso4217:USD",
}
]
},
},
)
fs._tags["us-gaap_Cash"] = {
"label": "Cash and equivalents",
"crdr": "debit",
"name": "Cash",
}
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": [
"Net total",
"Cash and equivalents",
],
"2024-08-31 -- 12 Months Ended": [33, 200],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
assert data[data.label == "Net total"].tag.tolist() == ["srt_AlphaXxxx"]
def test_apply_label_returns_none_for_none_and_nan_labels():
from pandas import DataFrame
fs = _fix_tag_fs(
tags={
"us-gaap_Revenue": {
"label": "Revenue",
"crdr": "credit",
"name": "Revenue",
}
},
instance={
"us-gaap_Revenue": {
"context": [
{
"value": "3",
"context_ref": "duration_2023_09_01_to_2024_08_31",
"unit": "iso4217:USD",
}
]
}
},
)
def _from_url(url, is_equity=False):
table = DataFrame(
{
"Statement of Operations $ in Millions": [None, "nan", "Revenue"],
"2024-08-31 -- 12 Months Ended": [1, 2, 3],
}
)
return table, DataFrame()
with patch.object(fs, "_download_statement_from_url", side_effect=_from_url):
data, _ = fs._download_statement("income")
assert data[data.label.isna()].iloc[0].tag is None
assert data[data.label == "nan"].iloc[0].tag is None
def test_download_statement_from_url_parent_tag_from_meta():
"""A meta tag carrying a parent_tag is mapped into the item map's parent_tag column."""
from pandas import DataFrame
fs = make_fs(
_document_type="10-Q",
_instance={
"us-gaap_AssetsCurrent": {
"context": [{"unit": "iso4217:USD", "decimals": "-6"}]
}
},
_tags={
"us-gaap_AssetsCurrent": {
"name": "AssetsCurrent",
"parent_tag": "us-gaap_Assets",
}
},
)
table = DataFrame(
{
"Balance Sheet $ in Millions": ["Total current assets"],
"Aug. 31, 2024 Ended": ["100"],
}
)
meta = DataFrame(
{0: ["Name", "Balance Type"], 1: ["us-gaap_AssetsCurrent", "debit"]}
)
with patch.object(fs, "_get_document", return_value=[table, meta]):
_, out_meta = fs._download_statement_from_url("https://sec.gov/R2.htm")
if not out_meta.empty and "parent_tag" in out_meta.columns:
assert "us-gaap_Assets" in set(out_meta.parent_tag.dropna())
def test_download_statement_from_url_empty_date_header():
"""An empty value-column header formats to None in the period-ending label."""
from pandas import DataFrame
fs = make_fs(_document_type="10-Q", _instance={}, _tags={})
table = DataFrame(
{
"Income Statement $ in Millions": ["Net sales and revenues"],
"": ["100"],
}
)
meta = DataFrame({0: ["Name", "Balance Type"], 1: ["us-gaap_Revenues", "credit"]})
with patch.object(fs, "_get_document", return_value=[table, meta]):
out_table, _ = fs._download_statement_from_url("https://sec.gov/R4.htm")
assert str(out_table.columns[1]).startswith("None -- ")
def test_download_statement_from_url_unparseable_date_header():
from pandas import DataFrame
fs = make_fs(_document_type="10-Q", _instance={}, _tags={})
table = DataFrame(
{
"Income Statement $ in Millions": ["Net sales and revenues"],
"Not a Date Header": ["100"],
}
)
meta = DataFrame({0: ["Name", "Balance Type"], 1: ["us-gaap_Revenues", "credit"]})
with patch.object(fs, "_get_document", return_value=[table, meta]):
out_table, _ = fs._download_statement_from_url("https://sec.gov/R4.htm")
assert str(out_table.columns[1]).startswith("None -- ")
def test_exhibit_title_mapping():
from openbb_sec.models.sec_financials import _exhibit_title
assert _exhibit_title("EX-21") == "Subsidiaries of the Registrant"
assert _exhibit_title("EX-23.1") == "Consents of Experts and Counsel"
assert _exhibit_title("EX-FOO") == ""
def test_parse_exhibit_index_variants():
html = (
'<a style="-sec-extract:exhibit" href="ex1.htm">Full Description Here</a>'
'<a style="-sec-extract:exhibit" href="ex1.htm">dup</a>'
'<a style="-sec-extract:exhibit" href="ex2.htm">.</a>'
'<a style="other" href="ex3.htm">Not an exhibit</a>'
'<a style="-sec-extract:exhibit">No href</a>'
)
fs = make_fs()
with (
patch.object(fs, "get_main_document_content", return_value=html),
patch.object(
fs, "get_main_document_url", return_value="https://sec.gov/x/m.htm"
),
):
assert fs._parse_exhibit_index() == [
{"value": "https://sec.gov/x/ex1.htm", "label": "Full Description Here"}
]
def test_parse_exhibit_index_with_row_number():
html = (
"<table><tr><td>10.1*</td><td>"
'<a style="-sec-extract:exhibit" href="ex10-1.htm">Stock Plan</a>'
"</td></tr></table>"
)
fs = make_fs()
with (
patch.object(fs, "get_main_document_content", return_value=html),
patch.object(
fs, "get_main_document_url", return_value="https://sec.gov/x/m.htm"
),
):
assert fs._parse_exhibit_index() == [
{"value": "https://sec.gov/x/ex10-1.htm", "label": "10.1 - Stock Plan"}
]
def test_parse_exhibit_index_prefers_longer_row_description():
"""The full row text wins when the anchor text is only a fragment."""
html = (
"<table><tr><td>10.2</td><td>"
'<a style="-sec-extract:exhibit" href="ex10-2.htm">A</a>'
" Material Definitive Agreement</td></tr></table>"
)
fs = make_fs()
with (
patch.object(fs, "get_main_document_content", return_value=html),
patch.object(
fs, "get_main_document_url", return_value="https://sec.gov/x/m.htm"
),
):
assert fs._parse_exhibit_index() == [
{
"value": "https://sec.gov/x/ex10-2.htm",
"label": "10.2 - A Material Definitive Agreement",
}
]
def test_exhibit_choices_uses_index_when_present():
fs = make_fs()
with patch.object(
fs, "_parse_exhibit_index", return_value=[{"value": "u", "label": "10.1 - X"}]
):
assert fs.exhibit_choices() == [{"value": "u", "label": "10.1 - X"}]
def test_exhibit_choices_embedded_fallback():
fs = make_fs(
_document_urls=[
{"type": "10-K", "sequence": "1"},
{"type": "EX-21", "sequence": "2", "description": "EXHIBIT 21"},
{"type": "EX-101.SCH", "sequence": "3", "description": "XBRL"},
{"type": "EX-99.1", "sequence": "4", "description": "Press Release"},
{"type": "EX-77", "sequence": "5", "description": ""},
]
)
with patch.object(fs, "_parse_exhibit_index", return_value=[]):
choices = fs.exhibit_choices()
values = {c["value"]: c["label"] for c in choices}
assert values["EX-21"] == "EX-21 - Subsidiaries of the Registrant"
assert values["EX-99.1"] == "EX-99.1 - Press Release"
assert values["EX-77"] == "EX-77"
assert not any("EX-101" in v for v in values)
def test_get_exhibit_by_url_html():
fs = make_fs(_use_cache=True)
with (
patch.object(fs, "_get_document", return_value="<html><body>x</body></html>"),
patch.object(fs, "_clean_html_to_text", return_value="Body markdown"),
):
assert fs.get_exhibit("https://sec.gov/x/ex.htm") == "Body markdown"
def test_get_exhibit_by_type_plain_text_reflow():
fs = make_fs(
_use_cache=True,
_document_urls=[{"type": "EX-21", "content": " SUBSIDIARIES\nApple Inc."}],
)
with patch.object(
fs, "_clean_html_to_text", return_value=" SUBSIDIARIES\nApple Inc."
):
assert "SUBSIDIARIES" in fs.get_exhibit("EX-21")
def test_get_exhibit_strips_document_wrapper():
fs = make_fs(_use_cache=True)
wrapped = "<DOCUMENT>\n<TYPE>EX-10.1\n<TEXT>\n<html><body>Real body</body></html>"
with (
patch.object(fs, "_get_document", return_value=wrapped),
patch.object(fs, "_clean_html_to_text", side_effect=lambda c, **k: c),
):
result = fs.get_exhibit("https://sec.gov/x/ex.htm")
assert "<TYPE>" not in result
assert "Real body" in result
def test_get_exhibit_not_found_and_empty():
fs = make_fs(_use_cache=True, _document_urls=[{"type": "EX-21"}])
assert fs.get_exhibit("EX-99") is None
with patch.object(fs, "_get_document", return_value=""):
assert fs.get_exhibit("EX-21") is None
def test_legacy_risk_factors_extracts_factors_section():
fs = make_fs()
mda = {
"text": "MD&A intro.\n\nFactors That May Affect Future Results and "
"Financial Condition\n\nCompetition is intense."
}
with patch.object(
fs, "get_item", side_effect=lambda *n: mda if n[0] == "7" else None
):
result = fs._legacy_risk_factors()
assert (
result[0]["risk_factor"]
== "Factors That May Affect Future Results and Financial Condition"
)
assert "Competition is intense" in result[0]["text"]
def test_legacy_risk_factors_empty_body():
fs = make_fs()
mda = {
"text": "MD&A.\n\nFactors That May Affect Future Results and "
"Financial Condition"
}
with patch.object(
fs, "get_item", side_effect=lambda *n: mda if n[0] == "7" else None
):
assert fs._legacy_risk_factors() == []
def test_legacy_risk_factors_no_section():
fs = make_fs()
with (
patch.object(fs, "get_item", return_value=None),
patch.object(fs, "_item_by_name", return_value=None),
):
assert fs._legacy_risk_factors() == []
def test_parse_exhibit_index_no_anchors():
fs = make_fs()
with patch.object(
fs, "get_main_document_content", return_value="<html>no exhibits here</html>"
):
assert fs._parse_exhibit_index() == []
def test_get_exhibit_decodes_bytes():
fs = make_fs(_use_cache=True, _document_urls=[{"type": "EX-21", "url": "u"}])
with (
patch.object(
fs, "_get_document", return_value=b"<html><body>Bytes body</body></html>"
),
patch.object(fs, "_clean_html_to_text", side_effect=lambda c, **k: c),
):
assert "Bytes body" in fs.get_exhibit("EX-21")