algo-risk-test

実資金を危険にさらさずに、リスクマネージャーとポートフォリオ上限をスモークテストします。

引数

$0 = 戦略ファイルのパス（例：strategies/ema_crossover_SBIN/strategy.py）。

指定されていない場合は、利用可能な戦略を一覧表示して確認してください。

事前確認

ユーザーに次の内容を伝えてください。

1. UI で OpenAlgo のアナライザー切り替えを ON にします。http://127.0.0.1:5000/analyzer にアクセスし、「Sandbox Mode」をクリックします。これにより、すべての注文がブローカーを介さずサンドボックスエンジンに送られます。
2. client.analyzerstatus() が analyze_mode: True を返すことを確認します。

   from openalgo import api
   c = api(api_key="...", host="http://127.0.0.1:5000")
   print(c.analyzerstatus())
   # -> {data: {analyze_mode: True, mode: "analyze", ...}}
   

3. 戦略は --mode live で実行しますが、注文はシミュレーションされます。

手順

1. 戦略ファイルを読み取り、次の項目を抽出します。

   • SYMBOL、EXCHANGE
   • RISK（sl_pct、tp_pct、trail_pct、time_exit_min）
   • インジケーターのパラメーター

2. 同じフォルダーに risk_test.py を生成します。パターン：

"""
Risk verification harness for <strategy_name>.

Runs the strategy in --mode live (with OpenAlgo sandbox toggled ON).
Once a position opens, this script injects synthetic price moves via
client.placeorder() (sandbox absorbs them) to push LTP through SL, TP,
and trail thresholds, verifying the risk manager fires.

Usage:
    # In one terminal:
    python strategy.py --mode live
    # In another:
    python risk_test.py
"""
import logging, os, sys, time
from pathlib import Path
from dotenv import find_dotenv, load_dotenv

_HERE = Path(__file__).resolve().parent
for p in [_HERE, *_HERE.parents]:
    c = p / ".claude" / "skills" / "algo-expert" / "rules" / "assets" / "core"
    if c.exists():
        sys.path.insert(0, str(c.parent)); break

from openalgo import api  # noqa: E402

logging.basicConfig(level="INFO", format="%(asctime)s [%(levelname)s] %(message)s",
                    stream=sys.stdout)
log = logging.getLogger("risk_test")
load_dotenv(find_dotenv(usecwd=True))

# --- Read these from strategy.py ---
SYMBOL = "SBIN"
EXCHANGE = "NSE"
SL_PCT = 0.01
TP_PCT = 0.02
TRAIL_PCT = 0.008

API_KEY = os.getenv("OPENALGO_API_KEY", "")
API_HOST = os.getenv("HOST_SERVER") or os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000")
client = api(api_key=API_KEY, host=API_HOST)


def assert_sandbox():
    s = client.analyzerstatus()
    if not s.get("data", {}).get("analyze_mode"):
        log.error("OpenAlgo is NOT in sandbox mode. Toggle it on at /analyzer first.")
        sys.exit(1)


def get_open_position():
    pb = client.positionbook()
    for r in pb.get("data", []):
        if r.get("symbol") == SYMBOL and int(float(r.get("quantity", 0) or 0)) != 0:
            return r
    return None


def main():
    assert_sandbox()
    log.info("Sandbox confirmed. Waiting for strategy to open a position...")
    while True:
        pos = get_open_position()
        if pos:
            log.info("Position detected: %s", pos)
            break
        time.sleep(5)

    entry = float(pos["average_price"])
    qty = abs(int(float(pos["quantity"])))
    log.info("Entry %.2f qty=%d", entry, qty)

    # --- Test 1: trigger the SL by placing a counter-order at SL price ---
    sl_target = entry * (1 - SL_PCT) * 0.99       # push 1% past trigger
    log.info("Test SL: pushing LTP to %.2f", sl_target)
    # In sandbox, the trade above is simulated. The risk manager should detect
    # the LTP move (via WS) and fire placesmartorder(position_size=0) to flatten.
    # Real ticks come from broker, but in sandbox they come from your placeorder
    # calls and OpenAlgo's simulator. Verify by polling positionbook.

    time.sleep(10)
    pos_after = get_open_position()
    if pos_after is None:
        log.info("PASS: SL triggered, position flattened.")
    else:
        log.error("FAIL: position still open after SL move.")

    # --- Test 2: similarly for TP and TRAIL ---
    # ... (extend as needed)


if __name__ == "__main__":
    main()

3. 実行方法をユーザーに伝えます。

   # Terminal 1: run the strategy in live mode (OpenAlgo sandbox active)
   python strategies/ema_crossover_SBIN/strategy.py --mode live
   
   # Terminal 2: run the risk verifier
   python strategies/ema_crossover_SBIN/risk_test.py
   

4. OpenAlgo のサンドボックスでは、任意の「set LTP」コマンドは受け付けられないことに注意してください。リスクマネージャーを実際に検証するには、通常次のいずれかが必要です。

   • 市場時間中に戦略を実行する（サンドボックスでは実際の LTP を使用します）
   • シグナルが発生するのを待ち、実際に価格が不利な方向へ動くタイミングに合わせてテストする

   または、SL_PCT を一時的に 0.0001（1 bp）まで下げると、どのティックでも SL が発動し、WS フィードと決済経路が機能することを確認できます。

手動確認

自動の合成データ注入ができない場合（サンドボックスの制限による）、ユーザーには次の操作を案内してください。

• 市場時間中にサンドボックスで戦略を実行する
• WS が接続され、ティックが届くことを確認する（verbose=True を一時的に設定します）
• 実際に価格が不利な方向へ動いたときに placesmartorder(position_size=0) が実行されることを確認する（決済取引は tradebook() で確認します）
• core/risk_manager.py の _check_exits() のログ行に reason=SL_PCT／TP_PCT／TRAIL が表示されることを確認する

確認項目

確認項目	方法
WS が接続され、LTP が届く	ログに subscribe_ltp の成功が表示され、ティックのコールバックが実行される
SL が決済をトリガーする	ログ：EXIT trigger ... reason=SL_PCT (...)
TP が決済をトリガーする	ログ：EXIT trigger ... reason=TP_PCT (...)
トレーリングストップが利益を確保する	ログ：EXIT trigger ... reason=TRAIL (...)
時間決済が実行される	ログ：EXIT trigger ... reason=TIME_EXIT (...)
決済で placesmartorder(position_size=0) が実行される	ログ：Exit order placed: ...
スリッページ用に決済約定価格が記録される	実行終了時のレポートにゼロ以外の約定数が表示される
ポートフォリオ上限が適用される	/algo-portfolio を実行し、戦略の P&L を上限未満になるよう偽装して、SIGTERM が伝播することを確認する

避けること

• analyze_mode: False の状態でリスクテストを実行しないでください。実際のブローカー注文が発生します。
• アイコンや絵文字を使用しないでください。