hawk-bt Python Library | Hawk-Backtester

システム全体構成を踏まえた hawk-bt の役割»

シミュレーション中、Python とブラウザは WebSocket で直結する。hawk-bt はその間に入り、接続管理・プロトコル変換・同期制御をすべて引き受ける。

Your Python

Strategy クラスを書く
step(ctx) で売買判断

hawk-bt（このライブラリ）

接続・再接続管理 / バイナリ RPC 変換
同期ポリシー適用 / ループ実行

WebSocket

Browser WASM

OHLC 保持・TP/SL 判定
ポジション管理・損益
レポート生成

Quick Start»

pip install hawk-bt

※ PyPI 上の hawk-backtester は本プロジェクトとは無関係の別パッケージです。

Strategy クラスを書く（下の例を参照）
ブラウザでシミュレーション画面を開く
python my_strategy.py を実行
ブラウザ側で「Start」を押す

セッションが終わると結果がログに出る。ブラウザが切断・再接続しても自動で次のセッションを受け付ける。

例: SMA クロスオーバー»

短期 SMA が長期 SMA を上抜けたら Buy、下抜けたら Sell。 TP/SL で自動決済。手動クローズはしない。

テンプレート（毎回ほぼ同じ）自由に書く部分（戦略ロジック）

from collections import deque
from hawk_bt import Strategy, Context, HawkEngine, configure_logging

class SMACross(Strategy):
    def __init__(self, short=10, long=30, tp=5.0, sl=3.0, units=10):
        self.short_p, self.long_p = short, long
        self.tp, self.sl, self.units = tp, sl, units
        self._hist: deque = deque(maxlen=long)
        self._prev = (None, None)

    def _sma(self, n):
        if len(self._hist) < n: return None
        return sum(list(self._hist)[-n:]) / n

    async def step(self, ctx: Context):  # ← 毎バー呼ばれるエントリポイントB
        price = ctx.state.snapshot.priceC
        self._hist.append(price)

        s, l = self._sma(self.short_p), self._sma(self.long_p)
        if s is None or l is None:
            self._prev = (s, l)
            return

        ps, pl = self._prev
        self._prev = (s, l)
        if ps is None: return

        # クロス検出
        if ps - pl <= 0 < s - l:
            await ctx.engine.place_ticket(side="buy", units=self.units,D
                                         take_profit=self.tp, stop_loss=self.sl)
        elif ps - pl >= 0 > s - l:
            await ctx.engine.place_ticket(side="sell", units=self.units,
                                         take_profit=self.tp, stop_loss=self.sl)

configure_logging(2)
HawkEngine().start(SMACross())A

コードの解説»

上の SMA クロスオーバーの例をもとに、hawk-bt がどう動くかを解説する。

シミュレーションループ

A HawkEngine().start(SMACross()) を実行すると、ブラウザからの接続を待ち受ける。接続が来るとバックテストが始まり、1 バーずつ以下の 3 ステップを繰り返す:

hawk-bt

WASM 市場シミュレーター

1get_snapshot()

受け取った snapshot を
ctx.state.snapshot に格納

情報提供

現在の市場状態を返す
balance, equity, price …

2strategy.step(ctx)

ユーザーの戦略を実行
売買判断 → 注文送信

注文指示

注文を受付・約定

3step_next()

進行指示

次のバーへ進む
TP/SL 判定・約定処理

step() の中身

B async def step(self, ctx: Context) は毎バー呼ばれるエントリポイント。この中で価格を読み、判断し、注文を出す。

C ctx.state.snapshot.price — 現在価格を取得して SMA を計算
D place_ticket(side="buy", ...) — ゴールデンクロス検出時に成行注文
D take_profit=self.tp, stop_loss=self.sl — TP/SL を絶対価格差で指定

同期ポリシー: eager vs deferred

D place_ticket() を呼ぶと WASM 側の状態が変わる。「その変化をいつ Python 側に反映するか」を制御するのが同期ポリシー。

deferred（デフォルト）

step(ctx): snap = ctx.state.snapshot ← ステップ頭の値 place_ticket("buy", 10) └ sync なし snap は変わらない place_ticket("sell", 5) └ sync なし snap はまだ古い step_next() ← ここでまとめて sync └ 次の get_snapshot() で tickets_num: 0 → 2 が反映

eager（UI「高精度」ボタン ON）

step(ctx): snap = ctx.state.snapshot ← 最新 place_ticket("buy", 10) └ sync ← 即座に snapshot 更新 tickets_num: 0 → 1 place_ticket("sell", 5) └ sync ← また更新 tickets_num: 1 → 2 tickets = get_ticket_list() └ sync ← 2件とも見える

deferred がデフォルト。ステップ境界でだけ同期するので高速。1 ステップ内で複数の注文を出す戦略に向く。ただし同一ステップ内では snapshot が古いまま。
eager は操作のたびに WASM と同期するので、常に最新の状態が見える。安全だが通信回数が多い。

同期ポリシーはブラウザ UI の「高精度」ボタンで切り替える。hawk-bt は接続時にこの設定を自動取得するので、コード側での指定は不要。

TP/SL 判定のタイミング

D take_profit=self.tp で設定した TP/SL は、注文を出したバーでは評価されない。 step_next() で次のバーに進んだとき、そのバーの高値・安値で判定される。

判定と約定価格は異なる。TP/SL のヒット判定にはバーの high / low を使うが、決済価格（約定価格）はそのバーの close が使われる。同一バーで TP と SL の両方がヒットした場合は不利側（SL）が優先される。

API — 情報取得»

ctx.state.snapshot

毎ステップ冒頭で自動更新される口座・市場の状態。呼び出し不要。

s = ctx.state.snapshot
s.price          # 現在価格
s.balance        # 口座残高
s.equity         # 評価額 (残高 + 含み損益)
s.margin_level   # 証拠金維持率
s.tickets_num    # オープンポジション数
s.step           # 現在のバー番号 (0-indexed)

Field	Type	Description
balance	float	口座残高
equity	float	評価額 (残高 + 含み損益)
used_margin	float	使用中証拠金
margin_level	float	証拠金維持率
price	float	現在価格
timestamp_ms	float	タイムスタンプ (ms)
step	int	バー番号 (0-indexed)
pending_orders	float	未約定注文数
tickets_num	float	オープンポジション数
ticket_all_num	float	全ポジション数 (クローズ含む)
bar_count	int	処理済みバー数
ticket_stat_0..3	float	ポジション統計
ticket_long_count	float	ロングポジション数
ticket_short_count	float	ショートポジション数
pending_long_count	float	ロング未約定注文数
pending_short_count	float	ショート未約定注文数
total_steps	int	シミュレーション総バー数

ctx.state.candles

全期間の OHLC + タイムスタンプ。セッション開始時に一括取得済み。以降の通信は発生しない。numpy 配列なのでスライスや演算がそのまま使える。

c = ctx.state.candles
i = ctx.state.snapshot.step

c.close[i]                      # 現在価格 (== snapshot.price)
c.close[i-1]                   # 前バーの終値
c.high[:i].max()                # ここまでの最高値
c.close[i-20:i].mean()          # 20期間 SMA

candles には全期間のデータが格納されている。 インジケーター計算には必ず candles.close[:step] のようにスライスし、現在バーより先のデータを参照しないこと。未来の価格を使った戦略はバックテスト結果を無効にする。

Field	Description
time	タイムスタンプ
open	始値
high	高値
low	安値
close	終値

情報取得 await ctx.engine.get_ticket_list()

オープンポジション一覧。呼ぶたびに通信が発生する。

Returns

np.ndarray — 各行が 1 ポジション

情報取得 await ctx.engine.fetch_events()

前回取得以降のイベント（決済、マージンコール等）。ターミナル状態の検出も内部で行う。

Returns

np.ndarray — shape (N, 5)

情報取得 await ctx.engine.refresh()

スナップショットだけ手動で再取得。イベントは取らない。

API — 注文・決済»

売買指示 await ctx.engine.place_ticket(side, units, ...)

成行注文。現在の市場価格で即時約定。

Param	Type	Note
side	str	`"buy"` / `"sell"`
units	int	ロット数 (1 以上)
take_profit	float?	TP 距離 (絶対値)
stop_loss	float?	SL 距離 (絶対値)
trailing_stop	float?	トレーリングストップ距離

TP/SL は絶対価格差で指定する。take_profit=5.0 なら「エントリーから 5.0 離れたら利確」。パーセンテージではない。

売買指示 await ctx.engine.place_order(side, order_type, price, units, ...)

指値 ("limit") / 逆指値 ("stop") 注文。指定価格に到達したら約定。

Param	Type	Note
side	str	`"buy"` / `"sell"`
order_type	str	`"limit"` / `"stop"`
price	float	注文価格
units	int	ロット数
take_profit	float?	TP 距離
stop_loss	float?	SL 距離
trailing_stop	float?	トレーリングストップ距離
time_limit	float?	有効期限 (バー数)

売買指示 await ctx.engine.close_positions(position_ids, actions, ratios)

ポジション決済。全決済でも一部決済でも。

Param	Type	Note
position_ids	list[int]	対象ポジションの ID
actions	list[int]	アクションコード（下表参照）
ratios	list[float]	決済比率 (0.0~1.0)。`action=2` のときのみ使用

Action codes

Code	動作	ratios の扱い
1	全決済	無視される
2	比率決済	`1.0` = 全決済、`0.5` = 半分決済

action=1 はポジションを即座に全決済する。action=2, ratio=1.0 でも全決済になるが、action=1 の方がシンプル。

制御 await ctx.engine.end_session()

シミュレーションを途中で打ち切る。通常は全バー消化か margin call で自動終了。

設定»

HawkEngine

Param	Default	Note
host	"127.0.0.1"	バインドアドレス
port	8787	ポート番号
on_result	None	結果コールバック (`BacktestResult` を受け取る)

BacktestResult

セッション終了後に on_result コールバックへ渡される結果オブジェクト。

Field / Method	Type	Description
steps	int	完了バー数
balance	ndarray	各バーの残高
equity	ndarray	各バーの評価額
price	ndarray	各バーの価格
total_orders	int	総注文数
win_count	int	勝ちトレード数
loss_count	int	負けトレード数
gross_profit	float	総利益（勝ちトレードの合計）
gross_loss	float	総損失（負けトレードの合計）
final_balance()	float	最終残高
max_drawdown()	float	最大 DD (例: -0.15)
max_drawdown_before_end()	float	最大 DD（残高 0 以前まで）

def on_done(r):
    print(f"{r.steps} bars, balance={r.final_balance():.2f}, DD={r.max_drawdown():.2%}")

HawkEngine(on_result=on_done).start(MyStrategy())

configure_logging(level)

Level	出力
1	WARNING のみ
2	INFO — 接続・切断・結果レポート
3	DEBUG — 注文ごとの詳細ログ

ctx.user

ステップ間でデータを持ち回すための辞書。インジケータのバッファ、トレード回数カウンタ、何でも入れられる。

async def step(self, ctx: Context):
    ctx.user.setdefault("n", 0)
    ctx.user["n"] += 1

参考: 独自インジケーターを使った戦略»

ctx.state.candles は全期間の OHLC を numpy 配列で保持しているため、現在のバー位置までスライスすれば任意のインジケーターをその場で計算できる。

import numpy as np
from hawk_bt import HawkEngine, Strategy, Context

class BollingerBreakout(Strategy):
    period = 20
    k      = 2.0

    async def step(self, ctx: Context):
        i = ctx.state.snapshot.step
        if i < self.period:
            return

        # 現在バーまでの終値をスライス
        close = ctx.state.candles.close[:i]

        # ボリンジャーバンドを計算
        sma   = close[-self.period:].mean()
        std   = close[-self.period:].std()
        upper = sma + self.k * std
        lower = sma - self.k * std

        price = ctx.state.snapshot.price

        if price > upper:
            await ctx.engine.place_ticket(
                side="buy", units=1,
                take_profit=std, stop_loss=std * 0.5,
            )
        elif price < lower:
            await ctx.engine.place_ticket(
                side="sell", units=1,
                take_profit=std, stop_loss=std * 0.5,
            )

HawkEngine().start(BollingerBreakout())

candles は numpy 配列なので、RSI・ATR・MACD など任意の指標を同じ要領で実装できる。

参考: 外部データとの突合»

OHLC 以外のデータ（出来高、ファンダメンタルズ等）を使いたい場合、自前で読み込んだデータを candles.time のタイムスタンプで突合すれば利用できる。

import pandas as pd
from hawk_bt import HawkEngine, Strategy, Context

class VolumeBreakout(Strategy):
    def __init__(self, volume_csv: str):
        # 外部の出来高データを読み込み、timestamp をキーにした Series を作成
        df = pd.read_csv(volume_csv)
        self.volume = df.set_index("timestamp")["volume"]

    async def step(self, ctx: Context):
        i = ctx.state.snapshot.step
        if i < 20:
            return

        # 現在バーの timestamp で出来高を取得
        ts = ctx.state.candles.time[i]
        vol = self.volume.get(ts, 0)

        # 直近20バーの平均出来高
        recent_ts = ctx.state.candles.time[i-20:i]
        avg_vol = self.volume.reindex(recent_ts).mean()

        # 出来高が平均の2倍以上 かつ 陽線 → エントリー
        if vol > avg_vol * 2 and ctx.state.snapshot.price > ctx.state.candles.open[i]:
            await ctx.engine.place_ticket(
                side="buy", units=1,
                take_profit=10.0, stop_loss=5.0,
            )

HawkEngine().start(VolumeBreakout("volume_data.csv"))

シミュレーターが提供するのは OHLC + タイムスタンプのみ。それ以外のデータは自由に持ち込める。

Security»

WebSocket 接続

hawk-bt の WS サーバーは 127.0.0.1:8787 にバインドされます。外部ネットワークには公開されません。
接続時に Origin ヘッダー検証を行い、許可リスト外のオリジンからの接続は 403 Forbidden で拒否されます。
Agent とブラウザが同一マシン上で実行されている必要があります。