axonx 0.0.2

Axon is a lightweight, extensible runner for ordered Python flows. It combines strict Pydantic configuration, shared runtime context, declared outputs, automatic provider discovery, and a small command-line interface. The package also includes reproducible Tushare market-data download flows with atomic Parquet snapshot persistence.

The PyPI distribution is named axonx; the Python package and command are both named axon. Axon runs flows but is not a task scheduler.

Capabilities

• Define a flow as an ordered or lazily generated sequence of zero-argument Python callables.
• Validate every flow configuration with strict Pydantic models that reject unknown fields.
• Share state through a flow context and require declared outputs before returning JSON.
• Discover installed provider plugins through the standard axon.flows entry-point group.
• Load provider modules directly for source checkouts through AXON_FLOW_MODULES.
• Download static, daily, stock-minute, ETF-minute, and overseas financial Tushare datasets.
• Persist deterministic Parquet snapshots with POSIX locks, exact comparison, atomic replacement, retained backups, and an optional local mirror.
• Reuse environment loading, rotating Loguru logging, DingTalk notifications, and a retrying paginated Tushare client.

Installation

Axon requires Python 3.11 or newer and supports Linux and macOS:

python -m pip install axonx

Quick start

List every built-in and installed-provider flow:

axon --list

Run the built-in demo flow:

axon --demo --x 1 --y 2

Axon writes the declared output as JSON:

{"result": 3}

The command format is axon --action --field value .... Hyphens in action and field names are normalized to underscores. Every configuration argument, including a boolean, must be passed as a --field value pair. Unknown, duplicate, missing, malformed, or extra arguments fail before the flow runs.

Define a flow

Configuration classes inherit from BaseConfig, and flows inherit from BaseFlow. Annotating config selects the configuration model. build_steps() returns or yields callables in execution order, while output_keys declares the context values that must exist after execution.

from collections.abc import Iterable

from axon.cli import BaseConfig, BaseFlow, Step, register


class GreetConfig(BaseConfig):
    name: str
    times: int = 1


@register("greet")
class GreetFlow(BaseFlow):
    config: GreetConfig
    output_keys = ("message",)

    def build_steps(self) -> Iterable[Step]:
        yield self.build_message

    def build_message(self) -> None:
        self.context["message"] = " ".join([f"hello {self.config.name}"] * self.config.times)

Steps share state through self.context. A generator-based build_steps() can use normal if, for, and yield from control flow, including decisions based on context written by an earlier step.

Add provider plugins

Keep provider flows in a separate Python distribution and expose the module containing their @register(...) decorators through the axon.flows entry-point group:

[project]
dependencies = ["axonx>=0.0.2,<0.1"]

[project.entry-points."axon.flows"]
example = "my_provider.flows"

After the provider is installed, Axon imports it automatically during discovery:

axon --list
axon --greet --name Axon --times 2

For a source checkout that is not installed, name one or more comma-separated importable modules and put their parent directories on PYTHONPATH:

AXON_FLOW_MODULES=my_provider.flows PYTHONPATH=/path/to/provider axon --list

Provider imports run after .env loading. Registration names are normalized, reserved names are rejected, and a duplicate action fails with the two conflicting classes identified.

Repository provider

This repository also contains the separately packaged proprietary axon-core provider. It uses the public axon.flows contract to add daily dataset assembly, feature engineering, model training and backtesting, online/offline prediction, and consistency checks without adding proprietary code to the axonx distribution.

For development from this repository, install the provider after the framework:

python -m pip install -e "./plugins/axon-core[dev]"
axon --list

The additional daily_* actions are owned and documented by axon-core; the actions below are the flows built into the public axonx package.

Built-in flows

Axon ships the following actions:

Action	Purpose
demo	Demonstrates ordered, conditional, and repeated steps by adding two integers.
download_tushare_static	Refreshes configured static and reference datasets.
download_tushare_daily	Downloads configured date-partitioned daily datasets.
download_tushare_stk_mins	Downloads stock minute bars from the persisted daily security universe.
download_tushare_etf_mins	Downloads ETF minute bars from persisted ETF reference data.
download_tushare_hk_financial	Downloads Hong Kong financial indicators by security.
download_tushare_us_financial	Downloads US financial indicators by security.

Daily and minute flows accept --start-date YYYYMMDD, --end-date YYYYMMDD, or --days-back N; financial flows accept explicit start and end dates. All download flows also accept --timeout, --retry-sleep-seconds, and --notify-dingtalk true|false. Minute and financial flows expose additional batching and progress controls through their strict configuration models.

Example:

axon --download-tushare-daily --start-date 20260101 --end-date 20260107

Each download returns a JSON summary containing selected APIs, date scope, elapsed time, and counts for attempted, created, changed, unchanged, rejected, and empty datasets. A completed run with rejected datasets exits with status 1; configuration and CLI errors exit with status 2.

Snapshot persistence

Tushare data is stored below <AXON_DATA_ROOT>/data/tushare; the default data root is axon_data. Before replacing a changed snapshot, Axon writes and syncs a temporary Parquet file, compares table structure and content exactly, and preserves the previous snapshot. The current file plus retained history is capped at seven versions.

Writes use an adjacent POSIX file lock, so snapshot persistence targets Linux and macOS. Setting AXON_MIRROR_PATH copies current snapshots and retained backups to the same relative path under an optional local mirror root.

Public utilities

The following functions and classes are exported from axon.utils:

API	Purpose
load_env(path=None, *, override=True)	Loads an explicit .env or the nearest one within five parent directories.
get_logger()	Returns the shared INFO logger with stderr and daily rotating file sinks.
send_dingtalk_message(title, text, msgtype="markdown", timeout=10.0)	Sends Markdown or text to configured DingTalk groups.
TushareClient(...)	Queries Tushare Pro with timeout, retry, pagination, overlap, and deduplication support.

Environment variables

The CLI calls load_env() before loading built-in flows or provider plugins.

Variable	Default	Purpose
AXON_DATA_ROOT	axon_data	Root for Axon-managed data.
AXON_MIRROR_PATH	Empty	Optional mirror root for Tushare snapshots and backups.
AXON_LOG_DIR	logs	Log file directory.
AXON_FLOW_MODULES	Empty	Comma-separated provider modules to import in addition to installed entry points.
AXON_TUSHARE_BASE_URL	http://api.waditu.com/dataapi	Tushare Pro endpoint or proxy URL.
AXON_TUSHARE_TOKEN	Empty	Tushare access token.
DINGTALK_CLIENT_ID	None	DingTalk application client ID.
DINGTALK_CLIENT_SECRET	None	DingTalk application client secret.
DINGTALK_CONVERSATIONS	None	JSON object mapping labels to group conversation IDs.

Network downloads and notifications require the corresponding credentials and are never performed by installation.

Development

python -m pip install -e ".[dev]"
axon --help
axon --demo --x 1 --y 2
pre-commit run --all-files
python -m build

License

axonx is licensed under the Apache License 2.0.