楽天証券の口座でデイトレの自動売買に挑戦しようと Windows / Excel 上で動作する マーケットスピード II RSS を利用した Python アプリ (Kabuto) を開発しています。
自動売買の実現は時間がかかると考えて、短周期移動平均線 MA1 と 出来高加重平均線 VWAP のクロス・シグナルを基本的な売買タイミングにして、セミオートで運用を始めています。
株価データ・取引の流れ（Windows 11）
毎日の取引をレビューするプログラムや取得したティックデータでバックテストをする体制が整いつつある状況にあります。有効に利用できるインジケータの探索についても少しずつ始めています。
日中のデイトレ以外のデータ収集や、定型分析をする活動もルーチン化してきたように見えます。
このようなアプローチを続ければ、この先に明るいゴールが待っているのでしょうか？
テクニカル・アナリストとしての経験を積んでいない自分でも、明るいゴールへたどり着ける可能性はゼロではないでしょうが、きっと時間がかかることでしょう。
だからバックアップ・プラン（プランＢ）も並行して進める必要があります。
強化学習という選択肢
昨年は、生成 AI の助けを借りながらデイトレに利用できる強化学習モデルの開発に取り組んできました。しかしモデルのチューニングにへこたれて一旦開発を中断。目に見える成果を求めてインジケータ寄りのアプローチへ移行して現在に至っています。
そろそろ強化学習モデルの開発を再開できる時期なのでしょう。平日は従来どおりのデイトレとデータ分析・バックテストを続け、週末は強化学習モデルの開発を少しづつ進めることにしました。
超シンプルな構成から
強化学習システムの基本構成は昨年と同じです。
強化学習システムの概要
昨年は、有効な観測値を増やすことやスケーリング、取引回数の制約などにとらわれてしまって、肝心の「そこそこ使えるモデル」の開発までたどり着けませんでした。
新たな Python プロジェクト HoneyBee を用意して、基本的な強化学習モデルの評価をできるようにしました。
再開にあたって下記のシンプルな問題を設定して、ここから肉付けをしていくことにします。
- 問題
- 報酬の最大化
- 報酬と収益が概ね比例するように考慮
- 状態 [Observation]
- Price（株価）
- Profit（含み損益）
- 行動 [Action]
- HOLD（何もしない）
- BUY（買建または返済）
- SELL（売建または返済）
- 返済ロジックは環境側で制御
- ナンピン禁止は行動マスクで処理
- 報酬 [Reward]
- 建玉時、含み損益の一定割合を付与
- 返済時、直前の含み損益を付与
- 約定コスト : 建玉、返済時いずれも約定コスト（ペナルティ）を引く
- 終了条件
- terminated
- なし
- truncated
- ティックデータが最終行に達した時
- 終了時、建玉があれば強制返済。報酬条件、約定コストは同じ。
約定コストをかけてエントリした後に、含み損益から一部が報酬（ペナルイティ）になり、建玉の返済に応じて、含み損益 - 約定コスト が報酬（ペナルティ）になるという単純な構成です。
建玉を持っていない HOLD 状態では報酬無し。エントリした状況に応じて報酬を付与することはまだしていません。
環境クラス
Google Gemini に Gymnasium を用いたカスタム環境の雛形を作ってもらって、それをベースにマスク対応（action_masks メソッド）とポジション管理の機能（PositionManager クラスのインスタンス）を加えた学習環境を作成しました。
また、観測値空間 observation_space は、与えられたティックデータの株価の 最大値、最小値 を High/Low に設定、含み損益の範囲は、±株価のレンジ を High/Low に設定しています。
以前、生成 AI に教えてもらったことによると、観測値空間 observation_space は、標準化あるいは規格化などでスケーリングした方が望ましいということでした。
しかし、去年はスケーリングすることに執着した結果、何を見ているかよくわからなくなってしまったので、ひとまずスケーリングは後回しにして、現実の株価レンジを適用したらどうなるかを確認したかったのです。
from typing import Any
import gymnasium as gym
import pandas as pd
from gymnasium import spaces
import numpy as np
from modules.posman import PositionManager
from structs.app_enum import ActionType, PositionType
class TrainingEnv(gym.Env):
# metadata defines render modes and framerate
metadata = {"render_modes": ["human", "rgb_array"], "render_fps": 30}
def __init__(self, code: str, df: pd.DataFrame, render_mode=None) -> None:
super().__init__()
self.df: pd.DataFrame = df
self.render_mode = render_mode
# 報酬関連
self.ratio_profit_hold = 0.1 # HOLD 時の含み損益から報酬
self.cost_contract = 1 # 約定手数料（スリッページ相当）
# ポジション・マネージャ
self.posman = posman = PositionManager()
posman.initPosition([code])
# インスタンス変数の初期化
self.code: str = code
self.row: int = 0
self.position: PositionType = PositionType.NONE
self.profit: float = 0.0
self.pnl_total: float = 0.0
# Define action_space（行動空間）
n_action_space = len(ActionType)
self.action_space = spaces.Discrete(n_action_space)
# Define observation_space（観測値空間）
"""
0. 株価［最小値, 最大値］
1. 含み益［-株価レンジ, +株価レンジ］
"""
price_max = df["Price"].max()
price_min = df["Price"].min()
price_rng = price_max - price_min
self.observation_space = spaces.Box(
low=np.array([price_min, -price_rng]),
high=np.array([price_max, +price_rng]),
shape=(2,),
dtype=np.float32
)
def action_masks(self) -> np.ndarray:
"""
行動マスク
【マスク】
- ナンピン取引の禁止
:return:
"""
if self.position == PositionType.NONE:
# 建玉なし → 取りうるアクション: HOLD, BUY, SELL
return np.array([1, 1, 1], dtype=np.int8)
elif self.position == PositionType.LONG:
# 建玉あり LONG → 取りうるアクション: HOLD, SELL
return np.array([1, 0, 1], dtype=np.int8)
elif self.position == PositionType.SHORT:
# 建玉あり SHORT → 取りうるアクション: HOLD, BUY
return np.array([1, 1, 0], dtype=np.int8)
else:
raise TypeError(f"Unknown PositionType: {self.position}")
def get_ts_price(self, row: int) -> tuple[float, float]:
"""
ティックデータから一行抽出
:param row:
:return:
"""
return self.df.iloc[row][["Time", "Price"]]
def get_transaction_result(self) -> pd.DataFrame:
"""
取引結果
:return:
"""
return self.posman.getTransactionResult()
def init_status(self) -> None:
"""
初期化処理
:return:
"""
self.row = 0
self.position = PositionType.NONE
self.profit: float = 0.0
self.pnl_total: float = 0.0
# ポジション・マネージャのリセットと初期化
self.posman.reset()
self.posman.initPosition([self.code])
def reset(self, seed=None, options=None) -> tuple[np.ndarray, dict[str, Any]]:
"""
環境のリセット処理
:param seed:
:param options:
:return:
"""
# Mandatory: seed the random number generator
super().reset(seed=seed)
# Initialize your state
_, price = self.get_ts_price(0)
profit = 0
observation = np.array([price, profit], dtype=np.float32)
info = {} # Additional debug info
self.init_status()
return observation, info
def step(self, action) -> tuple[np.ndarray, float, bool, bool, dict[str, Any]]:
"""
ステップ処理
:param action:
:return:
"""
# タイムスタンプと株価
ts, price = self.get_ts_price(self.row)
# 含み損益
profit = self.posman.getProfit(self.code, price)
# 観測値
observation = np.array([price, profit], dtype=np.float32)
# 報酬
reward = 0
# 建玉管理
action_type = ActionType(action)
if action_type == ActionType.BUY:
if self.position == PositionType.NONE:
# 建玉がなければ買建
self.posman.openPosition(self.code, ts, price, action_type)
self.position = PositionType.LONG # ポジションを更新
reward -= self.cost_contract # 約定コスト
elif self.position == PositionType.SHORT:
# 売建（ショート）であれば（買って）返済
self.posman.closePosition(self.code, ts, price)
self.position = PositionType.NONE # ポジションを更新
reward -= self.cost_contract # 約定コスト
reward += profit # 含み損益分そっくり報酬
else:
raise "trade rule violation!"
elif action_type == ActionType.SELL:
if self.position == PositionType.NONE:
# 建玉がなければ売建
self.posman.openPosition(self.code, ts, price, action_type)
self.position = PositionType.SHORT # ポジションを更新
reward -= self.cost_contract # 約定コスト
elif self.position == PositionType.LONG:
# 買建（ロング）であれば（売って）返済
self.posman.closePosition(self.code, ts, price)
self.position = PositionType.NONE # ポジションを更新
reward -= self.cost_contract # 約定コスト
reward += profit # 含み損益分そっくり報酬
else:
raise "trade rule violation!"
elif action_type == ActionType.HOLD:
# 含み益があれば幾分かを報酬に
reward += profit * self.ratio_profit_hold
else:
raise f"unknown action type {action_type}!"
# エピソード終了判定
terminated = False # Task finished (e.g., goal reached)
truncated = False # Time limit reached
info = {}
if len(self.df) - 1 <= self.row:
if self.posman.hasPosition(self.code):
reward -= self.cost_contract # 約定コスト
reward += profit * (1 - self.ratio_profit_hold) # 残りの含み損益分
self.posman.closePosition(self.code, ts, price, "強制返済")
self.position = PositionType.NONE # ポジションを更新
truncated = True # ← ステップ数上限による終了
info["done_reason"] = "truncated: last_tick"
self.row += 1
return observation, reward, terminated, truncated, info
def render(self) -> None:
# Implement visualization logic based on self.render_mode
pass
def close(self) -> None:
# Cleanup resources (e.g., close windows)
pass
エージェント
ある日のティックデータを用いてモデルに学習させ、同じティックデータから推論をするコードです。エピソードごとの報酬をログに取って、報酬トレンドを学習曲線としています。
import os
import pandas as pd
from sb3_contrib import MaskablePPO
from stable_baselines3.common.monitor import Monitor
from env import TrainingEnv
from funcs.io import get_sample_data
from funcs.plot import learning_curve
if __name__ == "__main__":
file_csv: str = "20260401_9984.csv"
code, df = get_sample_data(file_csv)
timesteps = 1_000_000
# ログフォルダの準備
dir_log = "./logs/"
os.makedirs(dir_log, exist_ok=True)
# 学習環境の準備
env0 = TrainingEnv(code, df)
env = Monitor(env0, dir_log) # Monitorの利用
# モデルの準備
model = MaskablePPO("MlpPolicy", env, verbose=1)
model.learn(total_timesteps=timesteps)
# 環境のリセット
obs, info = env.reset()
print(f"Initial observation: {obs}")
episode_over = False
total_reward = 0
# 推論の実行
print("Begin inference...")
counter = 0
while not episode_over:
# ラッピング前の閑居いうインスタンスから行動マスクの取得
action_masks = env0.action_masks()
# マスク情報付きで推論
action, _ = model.predict(obs, action_masks=action_masks)
obs, reward, terminated, truncated, info = env.step(action)
total_reward += reward
episode_over = terminated or truncated
# 結果の表示
print(f"Episode finished!\nTotal reward: {total_reward}")
df = env0.get_transaction_result()
print(df)
print(f"損益 : {df['損益'].sum()} 円, 約定係数 : {len(df)} 回")
# 環境の終了処理
env.close()
# 報酬トレンド/学習曲線
name_log = "monitor.csv"
# 最初の行の読み込みを除外
df_monitor = pd.read_csv(os.path.join(dir_log, name_log), skiprows=[0])
learning_curve(df_monitor, file_csv)
エージェントの実行例
100 回程度のエピソードを繰り返して、報酬トレンド（= 学習曲線）がどのようになるかを確認しました。
実行例
... （省略） ... ----------------------------------------- | rollout/ | | | ep_len_mean | 9.42e+03 | | ep_rew_mean | 4.55e+04 | | time/ | | | fps | 1032 | | iterations | 489 | | time_elapsed | 969 | | total_timesteps | 1001472 | | train/ | | | approx_kl | 0.000142657 | | clip_fraction | 0.000391 | | clip_range | 0.2 | | entropy_loss | -0.00388 | | explained_variance | -1.19e-07 | | learning_rate | 0.0003 | | loss | 1.75e+03 | | n_updates | 4880 | | policy_gradient_loss | -0.000139 | | value_loss | 3.89e+03 | ----------------------------------------- Initial observation: [3816. 0.] Begin inference... Episode finished! Total reward: 63549.800000000534 注文番号 注文日時 銘柄コード 売買 約定単価 約定数量 損益 備考 0 1 2026-04-01 09:10:27.467659950 9984 売建 3812.0 1 NaN 1 2 2026-04-01 15:24:48.622090101 9984 買埋 3759.0 1 53.0 強制返済 損益 : 53.0 円, 約定係数 : 2 回
損益は 1 株単位です。
エピソード毎の報酬トレンド（学習曲線）を以下のようになりました。なお、この例はそのうちの比較的良好な結果の一つにすぎません。追試で繰り返しましたが同じような傾向になりませんでした。まだまだ設計が甘いのでしょう。
エピソード毎の報酬トレンド（学習曲線）
少なくともこの例では、報酬トレンドを見る限り、こんな単純な設定でも何となく右肩上がりで学習が進んでいます。
学習中の収益や約定回数をログに残せるようにして、評価を続ける予定です。
Google Gemini による評価
学習中に膨大なログが吐き出されるので、なにも確認しないのは勿体ないので、Google Gemini に評価してもらいました（抜粋）。
添付の強化学習 (SB3 の MaskablePPO) の出力を評価してもらえますか？
添付されたMaskablePPO（Stable Baselines3）の学習ログを確認しました。約100万タイムステップ（1,001,472 steps）までの学習経過について、主要な指標に基づき評価します。
1. 学習の進捗と収束性（ep_rew_mean）
最もポジティブな点は、報酬の平均値（ep_rew_mean）が劇的に改善していることです。
- 学習初期（約1万ステップ付近）では -3,970 程度でしたが 、15万ステップ付近でプラスに転じ 、最終的には 45,500 まで上昇しています。
- 報酬が右肩上がりに伸びており、エージェントが環境のルールを適切に学習し、目標とする行動を獲得できていることを示しています。
2. 方策の確定度（entropy_loss）
- entropy_loss が学習初期の -0.85 から 、終盤には -0.00388 付近まで絶対値が低下しています。
- これは方策（Policy）が非常に確定的なもの（Deterministic）になっていることを意味します。エージェントが「どの状態でどの行動をとるべきか」に迷いがなくなっている状態ですが、エントロピーが低すぎると新しい状況への適応能力が下がる可能性があるため、環境にランダム性が高い場合は少し注意が必要です。
3. 価値関数の精度（explained_variance）
- explained_variance の値が 0 または マイナスの微小値（-1.19e-07など）で推移しています。
- これは、クリティック（価値関数ネットワーク）が報酬の予測をうまく行えていないことを示唆しています。強化学習において報酬のノイズが非常に大きい場合や、報酬が疎（Sparse）な場合に頻繁に見られる傾向です。
- 報酬は伸びているため学習自体は進んでいますが、価値関数の精度が低いと、今後さらに複雑な状況に対応させる際に学習が不安定になるリスクがあります。
総評と今後のアドバイス
全体として、「学習は非常に順調に進んでおり、期待した行動パターンを習得できている」と評価できます。次のステップとして、以下の検討をお勧めします。
- 検証（テスト）の実施: 学習データ以外の期間、あるいは異なるシード値でのテストを行い、過学習（オーバーフィッティング）が起きていないか確認してください。特にエントロピーがこれだけ低いと、特定のパターンに固執している可能性があります。
- 価値関数の改善: もし今後の学習が頭打ちになるようであれば、報酬の正規化（Scaling）や、ニューラルネットワークの層の深さ、特徴量（State）の再検討を行うことで、explained_variance を改善できる可能性があります。
- エピソード長の確認: 1エピソードが平均 9,420ステップ と比較的長めです。特定の局面での判断ミスが後半に大きく響く構造になっていないか、報酬設計を見直す余地があるかもしれません。
参考サイト
- マーケットスピード II RSS | 楽天証券のトレーディングツール
- マーケットスピード II RSS 関数マニュアル
- 注文 | マーケットスピード II RSS オンラインヘルプ | 楽天証券のトレーディングツール
- PythonでGUIを設計 | Qtの公式Pythonバインディング
- Python in Excel alternative: Open. Self-hosted. No limits.
- Book - xlwings Documentation
0 件のコメント:
コメントを投稿