安装与快速入门¶

#!/usr/bin/env python3
"""
AXON 快速入门示例

演示 TradingEnv 的核心 API：
- 创建环境（config + market_data + reward）
- reset() 获取初始观测
- step(action) 执行交易并返回 (obs, reward, terminated, truncated, info)
"""

import axon_quant

# ── 1. 准备合成行情数据 ────────────────────────────────
# 无需外部 CSV/Parquet，直接生成 100 根几何布朗运动 K 线
market_data = [
    {
        "timestamp": i,               # 时间戳（任意递增整数）
        "open": 100.0 + i * 0.1,      # 开盘价
        "high": 100.5 + i * 0.1,      # 最高价
        "low": 99.5 + i * 0.1,        # 最低价
        "close": 100.0 + i * 0.1,     # 收盘价
        "volume": 1000.0,             # 成交量
    }
    for i in range(100)
]

# ── 2. 创建交易环境 ────────────────────────────────────
env = axon_quant.rl.TradingEnv(
    # 环境配置：初始资金、交易成本、最大步数等
    config={
        "initial_capital": 100_000.0,   # 初始资金 10 万
        "transaction_cost": 0.001,      # 交易成本 10 bps
        "slippage": 0.0005,             # 滑点 5 bps
        "max_steps": 100,               # 每回合最多 100 步
        "seed": 42,                     # 随机种子（保证可复现）
        "symbol": "BTCUSDT",            # 交易标的
        "return_window": 20,            # 收益率历史窗口（用于 Sharpe 奖励）
    },
    # 动作空间：连续动作，单维目标仓位比例 [-1, 1]
    #   1.0  = 全仓做多
    #   0.0  = 空仓
    #   -1.0 = 全仓做空
    action_space={"type": "continuous", "min": -1.0, "max": 1.0},
    # 行情数据（必填）
    market_data=market_data,
    # 奖励函数："pnl"（盈亏）/ "sharpe"（夏普比率）/ "sortino"（索提诺比率）
    reward="pnl",
)

# ── 3. 重置环境，获取初始观测 ──────────────────────────
obs = env.reset()
print("初始观测特征:", obs["features"])       # 归一化后的 close + volume
print("特征名称:", obs["feature_names"])      # ["close", "volume"]

# ── 4. 执行一步交易 ────────────────────────────────────
# 动作：做多 50% 仓位（连续动作空间下传入 list[float]）
action = [0.5]
obs, reward, terminated, truncated, info = env.step(action)

print(f"\n执行 action={action} 后:")
print(f"  reward      = {reward:.4f}")               # 本步奖励（PnL 变化）
print(f"  portfolio   = {info['portfolio_value']:.2f}")  # 当前组合市值
print(f"  trades      = {info['trades_executed']}")       # 已成交笔数
print(f"  step        = {info['current_step']}")           # 当前步数
print(f"  done        = {info['done']}")                   # 回合是否结束

# ── 5. 渲染环境状态（ASCII 文本）────────────────────────
print("\n环境状态:", env.render())
# 输出示例: step=1/100 | value=$100000.00 | cash=$50000.00 | pos=0.5000 | trades=1 | cost=$50.00 | done=false

#!/usr/bin/env python3
"""
随机策略基线验证

目的：
1. 端到端冒烟测试 —— 验证 Rust 扩展 + Python 接口 + 行情数据流通畅
2. 性能基线 —— 后续训练的策略应明显优于随机策略
3. CI 入口 —— 无需 GPU / 训练依赖即可运行
"""

import random
import axon_quant

# 生成 500 根合成 K 线（带随机游走 + 高斯噪声）
def make_synthetic_data(n=500, seed=42):
    rng = random.Random(seed)
    price = 100.0
    bars = []
    for t in range(n):
        open_p = price
        ret = rng.gauss(0.0, 0.01)          # 日收益率 ~ N(0, 1%)
        close = max(1e-6, open_p * (1.0 + ret))
        spread = abs(close - open_p) + open_p * 0.005
        high = max(open_p, close) + spread * rng.random()
        low = max(1e-6, min(open_p, close) - spread * rng.random())
        volume = abs(1000.0 + 200.0 * rng.gauss(0.0, 1.0))
        bars.append({
            "timestamp": t, "open": open_p, "high": high,
            "low": low, "close": close, "volume": volume,
        })
        price = close
    return bars


market_data = make_synthetic_data(n=500, seed=42)

# 创建环境
env = axon_quant.rl.TradingEnv(
    config={"initial_capital": 100_000.0, "max_steps": 500, "seed": 42},
    market_data=market_data,
    reward="pnl",
)

# 运行 5 个随机 episode
n_episodes = 5
records = []
for ep in range(n_episodes):
    env.reset()
    total_reward = 0.0
    steps = 0
    done = False
    rng = random.Random(ep)

    while not done and steps < 500:
        action = [rng.uniform(-1.0, 1.0)]   # 随机生成目标仓位 [-1, 1]
        obs, reward, terminated, truncated, info = env.step(action)
        total_reward += reward
        steps += 1
        done = terminated or truncated

    records.append({
        "episode": ep,
        "steps": steps,
        "total_reward": total_reward,
        "final_value": info["portfolio_value"],
    })
    print(f"Episode {ep}: steps={steps}, reward={total_reward:.4f}, "
          f"final_value={info['portfolio_value']:,.2f}")

# 汇总统计
mean_reward = sum(r["total_reward"] for r in records) / n_episodes
mean_value = sum(r["final_value"] for r in records) / n_episodes
print(f"\n=== 随机策略基线汇总 ===")
print(f"平均奖励: {mean_reward:.4f}")
print(f"平均净值: {mean_value:,.2f}")

# 验收：所有 episode 至少跑过 5 步即视为通过
completed = sum(1 for r in records if r["steps"] >= 5)
assert completed == n_episodes, f"仅 {completed}/{n_episodes} episodes 正常完成"
print("✅ 验收通过：AXON 环境运行正常！")