レンジブレイク手法でのトレードをアシストするBotのサンプル

FXシステムトレードフレームワーク「Jiji」 のサンプルその3。
レンジブレイク手法を使ったトレードをアシストするBotを作ってみました。

FX Wroks さんのサイト に掲載されていた「レンジブレイクを狙うシンプルな順張り」手法を、そのままJijiに移植してみたものです。

動作

以下のような動作をします。

• 1) Botがレートを監視し、レンジブレイクをチェック
  • 条件は、サイトの内容と同等、8時間レートが100pips内で推移したあと、上or下に抜ける、としました。
  • 待つ期間やpipsは、パラメータで調整できるようにしています。
• 2) レンジブレイクを検出したら、スマホに通知を送信します
  • ダマしが多いので、今回は通知を送って判断する形に。
• 3) 通知を受けて最終判断を行い、トレードを実行。
  • 通知にあるボタンを押すことで、売or買で成行注文を実行できるようにしています。
  • 決済は、トレーリングストップで。

軽く動かしてみた感想

軽くテストしてみましたが、思ったよりもダマしに引っかかる感じですね。

これは、まぁまぁ。

これは、ブレイクと判定された時点で下げが終わっている・・。

これは、一度上にブレイクしたあと、逆方向に進んでいます・・・。

ブレイクの条件を調整してみる、移動平均でのトレンドチェックと組み合わせるなど、カスタマイズして使ってみてください。

コード

# === レンジブレイクでトレードを行うエージェント
class RangeBreakAgent

  include Jiji::Model::Agents::Agent

  def self.description
    <<-STR
レンジブレイクでトレードを行うエージェント。
 - 指定期間(デフォルトは8時間)のレートが一定のpipsに収まっている状態から、
   レンジを抜けたタイミングで通知を送信。
 - 通知からトレード可否を判断し、取引を実行できます。
 - 決済はトレーリングストップで行います。
    STR
  end

  # UIから設定可能なプロパティの一覧
  def self.property_infos
    [
      Property.new('target_pair',  '対象とする通貨ペア',      'USDJPY'),
      Property.new('range_period', 'レンジを判定する期間(分)',   60 * 8),
      Property.new('range_pips',    'レンジ相場とみなす値幅(pips)', 100),
      Property.new('trailing_stop_pips',
        'トレールストップで決済する値幅(pips)',                       30),
      Property.new('trade_units',   '取引数量',                      1)
    ]
  end

  def post_create
    pair = broker.pairs.find { |p| p.name == @target_pair.to_sym }
    @checker = RangeBreakChecker.new(
      pair, @range_period.to_i, @range_pips.to_i)
  end

  def next_tick(tick)
    # レンジブレイクしたかどうかチェック
    result = @checker.check_range_break(tick)
    # ブレイクしていたら通知を送る
    send_notification(result) if result[:state] != :no
  end

  def execute_action(action)
    case action
    when 'range_break_buy'  then buy
    when 'range_break_sell' then sell
    else '不明なアクションです'
    end
  end

  def state
    { checker: @checker.state }
  end

  def restore_state(state)
    @checker.restore_state(state[:checker]) if state[:checker]
  end

  private

  def sell
    broker.sell(@target_pair.to_sym, @trade_units.to_i, :market, {
      trailing_stop: @trailing_stop_pips.to_i
    })
    '売注文を実行しました'
  end

  def buy
    broker.buy(@target_pair.to_sym, @trade_units.to_i, :market, {
      trailing_stop: @trailing_stop_pips.to_i
    })
    '買注文を実行しました'
  end

  def send_notification(result)
    message = "#{@target_pair} #{result[:price]}" \
      + ' がレンジブレイクしました。取引しますか?'
    @notifier.push_notification(message, [create_action(result)])
    logger.info "#{message} #{result[:state]} #{result[:time]}"
  end

  def create_action(result)
    if result[:state] == :break_high
      { 'label'  => '買注文を実行', 'action' => 'range_break_buy' }
    else
      { 'label'  => '売注文を実行', 'action' => 'range_break_sell' }
    end
  end

end

class RangeBreakChecker

  def initialize(pair, period, range_pips)
    @pair       = pair
    @range_pips = range_pips
    @candles    = Candles.new(period * 60)
  end

  def check_range_break(tick)
    tick_value = tick[@pair.name]
    result = check_state(tick_value, tick.timestamp)
    @candles.reset unless result == :no
    # 一度ブレイクしたら、一旦状態をリセットして次のブレイクを待つ
    @candles.update(tick_value, tick.timestamp)
    {
      state: result,
      price: tick_value.bid,
      time:  tick.timestamp
    }
  end

  def state
    @candles.state
  end

  def restore_state(state)
    @candles.restore_state(state)
  end

  private

  # レンジブレイクしているかどうか判定する
  def check_state(tick_value, time)
    highest = @candles.highest
    lowest  = @candles.lowest
    return :no if highest.nil? || lowest.nil?
    return :no unless over_period?(time)

    diff = highest - lowest
    return :no if diff >= @range_pips * @pair.pip
    calculate_state( tick_value, highest, diff )
  end

  def calculate_state( tick_value, highest, diff )
    center = highest - diff / 2
    pips = @range_pips / 2 * @pair.pip
    if tick_value.bid >= center + pips
      return :break_high
    elsif tick_value.bid <= center - pips
      return :break_low
    end
    :no
  end

  def over_period?(time)
    oldest_time = @candles.oldest_time
    return false unless oldest_time
    (time.to_i - oldest_time.to_i) >= @candles.period
  end

end

class Candles

  attr_reader :period

  def initialize(period)
    @candles     = []
    @period      = period
    @next_update = nil
  end

  def update(tick_value, time)
    time = Candles.normalize_time(time)
    if @next_update.nil? || time > @next_update
      new_candle(tick_value, time)
    else
      @candles.last.update(tick_value, time)
    end
  end

  def highest
    high = @candles.max_by { |c| c.high }
    high.nil? ? nil : BigDecimal.new(high.high, 10)
  end

  def lowest
    low = @candles.min_by { |c| c.low }
    low.nil? ? nil : BigDecimal.new(low.low, 10)
  end

  def oldest_time
    oldest = @candles.min_by { |c| c.time }
    oldest.nil? ? nil : oldest.time
  end

  def reset
    @candles     = []
    @next_update = nil
  end

  def new_candle(tick_value, time)
    limit = time - period
    @candles = @candles.reject { |c| c.time < limit }

    @candles << Candle.new
    @candles.last.update(tick_value, time)

    @next_update = time + (60 * 5)
  end

  def state
    {
      candles:     @candles.map { |c| c.to_h },
      next_update: @next_update
    }
  end

  def restore_state(state)
    @candles = state[:candles].map { |s| Candle.from_h(s) }
    @next_update = state[:next_update]
  end

  def self.normalize_time(time)
    Time.at((time.to_i / (60 * 5)).floor * 60 * 5)
  end

end

class Candle

  attr_reader :high, :low, :time

  def initialize(high = nil, low = nil, time = nil)
    @high = high
    @low  = low
    @time = time
  end

  def update(tick_value, time)
    price = extract_price(tick_value)
    @high = price if @high.nil? || @high < price
    @low  = price if @low.nil?  || @low > price
    @time = time  if @time.nil?
  end

  def to_h
    { high: @high, low: @low, time: @time }
  end

  def self.from_h(hash)
    Candle.new(hash[:high], hash[:low], hash[:time])
  end

  private

  def extract_price(tick_value)
    tick_value.bid
  end

end

トラップリピートイフダンのような注文を発行するエージェントのサンプル

FXシステムトレードフレームワーク「Jiji」のサンプル その2、として、 トラップリピートイフダンのような注文を発行するエージェントを作ってみました。

※トラップリピートイフダン(トラリピ)は、マネースクウェアジャパン（M2J）の登録商標です。

トラップリピートイフダンとは

指値/逆指値の注文と決済を複数組み合わせて行い、その中でレートが上下することで利益を出すことを狙う、発注ロジックです。 具体的にどういった動きをするのかは、マネースクウェアジャパン のサイトがとてもわかりやすいので、そちらをご覧ください。

www.toraripifx.com

特徴

FX研究日記さんの評価記事が参考になります。

tasfx.net

• レンジ相場では、利益を出しやすい
• ×レートが逆行すると損失を貯めこんでしまう

仕組みからして、いわゆるコツコツドカンなシステムという印象です。 レンジ相場なら利益を積み上げやすいので、トレンドを判定するロジックと組み合わせて、レートが一定のレンジで動作しそうになったら稼働させる、などすれば使えるかも。

エージェントのコード

• 実装は、こちらのサイトで配布されているEAを参考にさせていただきました。
• TrapRepeatIfDoneAgentが、エージェントの本体です。これをバックテストやリアルトレードで動作させればOK。
  • エージェントファイルの追加の方法など、Jijiの基本的な使い方はこちらをご覧ください。
• 機能の再利用ができるように、発注処理はTrapRepeatIfDoneに実装しています。

# === トラップリピートイフダンのような注文を発行するエージェント
class TrapRepeatIfDoneAgent

  include Jiji::Model::Agents::Agent

  def self.description
    <<-STR
トラップリピートイフダンのような注文を発行するエージェント
      STR
  end

  # UIから設定可能なプロパティの一覧
  def self.property_infos
    [
      Property.new('trap_interval_pips', 'トラップを仕掛ける間隔(pips)', 50),
      Property.new('trade_units',        '1注文あたりの取引数量',         1),
      Property.new('profit_pips',        '利益を確定するpips',         100),
      Property.new('slippage',           '許容スリッページ(pips)',       3)
    ]
  end

  def post_create
    @trap_repeat_if_done = TrapRepeatIfDone.new(
      broker.pairs.find {|p| p.name == :USDJPY }, :buy,
      @trap_interval_pips.to_i,
      @trade_units.to_i, @profit_pips.to_i, @slippage.to_i, logger)
  end

  def next_tick(tick)
    @trap_repeat_if_done.register_orders(broker)
  end

  def state
    @trap_repeat_if_done.state
  end

  def restore_state(state)
    @trap_repeat_if_done.restore_state(state)
  end

end


# トラップリピートイフダンのような注文を発行するクラス
class TrapRepeatIfDone

  # コンストラクタ
  #
  # target_pair:: 現在の価格を格納するTick::Valueオブジェクト
  # sell_or_buy:: 取引モード。 :buy の場合、買い注文を発行する。 :sellの場合、売
  # trap_interval_pips:: トラップを仕掛ける間隔(pips)
  # trade_units:: 1注文あたりの取引数量
  # profit_pips:: 利益を確定するpips
  # slippage:: 許容スリッページ。nilの場合、指定しない
  def initialize(target_pair, sell_or_buy=:buy, trap_interval_pips=50,
    trade_units=1, profit_pips=100, slippage=3, logger=nil)

    @target_pair        = target_pair
    @trap_interval_pips = trap_interval_pips
    @slippage           = slippage

    @mode = if sell_or_buy == :sell
      Sell.new(target_pair, trade_units, profit_pips, slippage, logger)
    else
      Buy.new(target_pair, trade_units, profit_pips, slippage, logger)
    end

    @logger = logger

    @registerd_orders   = {}
  end

  # 注文を登録する
  #
  # broker:: broker
  def register_orders(broker)
    broker.instance_variable_get(:@broker).refresh_positions
    # 常に最新の建玉を取得して利用するようにする
    # TODO 公開APIにする

    each_traps(broker.tick) do |trap_open_price|
      next if order_or_position_exists?(trap_open_price, broker)
      register_order(trap_open_price, broker)
    end
  end

  def state
    @registerd_orders
  end

  def restore_state(state)
    @registerd_orders = state unless state.nil?
  end

  private

  def each_traps(tick)
    current_price = @mode.resolve_current_price(tick[@target_pair.name])
    base = resolve_base_price(current_price)
    6.times do |n| # baseを基準に、上下3つのトラップを仕掛ける
      trap_open_price = BigDecimal.new(base, 10) \
        + BigDecimal.new(@trap_interval_pips, 10) * (n-3) * @target_pair.pip
      yield trap_open_price
    end
  end

  # 現在価格をtrap_interval_pipsで丸めた価格を返す。
  #
  #  例) trap_interval_pipsが50の場合、
  #  resolve_base_price(120.10) # -> 120.00
  #  resolve_base_price(120.49) # -> 120.00
  #  resolve_base_price(120.51) # -> 120.50
  #
  def resolve_base_price(current_price)
    current_price = BigDecimal.new(current_price, 10)
    pip_precision = 1 / @target_pair.pip
    (current_price * pip_precision / @trap_interval_pips ).ceil \
      * @trap_interval_pips / pip_precision
  end

  # trap_open_priceに対応するオーダーを登録する
  def register_order(trap_open_price, broker)
    result = @mode.register_order(trap_open_price, broker)
    unless result.order_opened.nil?
      @registerd_orders[key_for(trap_open_price)] \
        = result.order_opened.internal_id
    end
  end

  # trap_open_priceに対応するオーダーを登録済みか評価する
  def order_or_position_exists?(trap_open_price, broker)
    order_exists?(trap_open_price, broker) \
    || position_exists?(trap_open_price, broker)
  end
  def order_exists?(trap_open_price, broker)
    key = key_for(trap_open_price)
    return false unless @registerd_orders.include? key
    id = @registerd_orders[key]
    order = broker.orders.find {|o| o.internal_id == id }
    return !order.nil?
  end
  def position_exists?(trap_open_price, broker)

    # trapのリミット付近でレートが上下して注文が大量に発注されないよう、
    # trapのリミット付近を開始値とする建玉が存在する間は、trapの注文を発行しない
    slipage_price = (@slippage.nil? ? 10 : @slippage) * @target_pair.pip
    position = broker.positions.find do |p|
      # 注文時に指定したpriceちょうどで約定しない場合を考慮して、
      # 指定したslippage(指定なしの場合は10pips)の誤差を考慮して存在判定をする
      p.entry_price < trap_open_price + slipage_price \
      && p.entry_price > trap_open_price - slipage_price
    end
    return !position.nil?
  end

  def key_for(trap_open_price)
    (trap_open_price * (1 / @target_pair.pip)).to_i.to_s
  end

  # 取引モード(売 or 買)
  # 買(Buy)の場合、買でオーダーを行う。売(Sell)の場合、売でオーダーを行う。
  class Mode

    def initialize(target_pair, trade_units, profit_pips, slippage, logger)
      @target_pair  = target_pair
      @trade_units  = trade_units
      @profit_pips  = profit_pips
      @slippage     = slippage
      @logger       = logger
    end

    # 現在価格を取得する(買の場合Askレート、売の場合Bidレートを使う)
    #
    # tick_value:: 現在の価格を格納するTick::Valueオブジェクト
    # 戻り値:: 現在価格
    def resolve_current_price(tick_value)
    end

    # 注文を登録する
    def register_order(trap_open_price, broker)
    end

    def calculate_price(price, pips)
      price = BigDecimal.new(price, 10)
      pips  = BigDecimal.new(pips,  10) * @target_pair.pip
      (price + pips).to_f
    end
    def pring_order_log(mode, options, timestamp)
      return unless @logger
      message = [
        mode, timestamp, options[:price], options[:take_profit],
        options[:lower_bound], options[:upper_bound]
      ].map {|item| item.to_s }.join(" ")
      @logger.info message
    end
  end

  class Sell < Mode
    def resolve_current_price(tick_value)
      tick_value.bid
    end
    def register_order(trap_open_price, broker)
      timestamp = broker.tick.timestamp
      options = create_option(trap_open_price, timestamp)
      pring_order_log("sell", options, timestamp)
      broker.sell(@target_pair.name, @trade_units, :marketIfTouched, options)
    end
    def create_option(trap_open_price, timestamp)
      options = {
        price:       trap_open_price.to_f,
        take_profit: calculate_price(trap_open_price, @profit_pips*-1),
        expiry:      timestamp + 60*60*24*7
      }
      unless @slippage.nil?
        options[:lower_bound] = calculate_price(trap_open_price, @slippage*-1)
        options[:upper_bound] = calculate_price(trap_open_price, @slippage)
      end
      options
    end
  end

  class Buy < Mode
    def resolve_current_price(tick_value)
      tick_value.ask
    end
    def register_order(trap_open_price, broker)
      timestamp = broker.tick.timestamp
      options = create_option(trap_open_price, timestamp)
      pring_order_log("buy", options, timestamp)
      broker.buy(@target_pair.name, @trade_units, :marketIfTouched, options)
    end
    def create_option(trap_open_price, timestamp)
      options = {
        price:       trap_open_price.to_f,
        take_profit: calculate_price(trap_open_price, @profit_pips),
        expiry:      timestamp + 60*60*24*7
      }
      unless @slippage.nil?
        options[:lower_bound] = calculate_price(trap_open_price, @slippage*-1)
        options[:upper_bound] = calculate_price(trap_open_price, @slippage)
      end
      options
    end
  end

end

インタラクティブにトレーリングストップ決済を行うBotを作ってみた

FXシステムトレードフレームワーク「Jiji」の使い方サンプル その1、ということで、 Jijiを使って、インタラクティブにトレーリングストップ決済を行うBotを作ってみました。

トレーリングストップとは

建玉(ポジション)の決済方法の一つで、「最高値を更新するごとに、逆指値の決済価格を切り上げていく」決済ロジックです。

例) USDJPY/120.10で買建玉を作成。これを、10 pips でトレーリングストップする場合、

• 建玉作成直後は、120.00 で逆指値決済される状態になる
• レートが 120.30 になった場合、逆指値の決済価格が高値に合わせて上昇し、120.20に切り上がる
• その後、レートが120.20 になると、逆指値で決済される

トレンドに乗っている間はそのまま利益を増やし、トレンドが変わって下げ始めたら決済する、という動きをする決済ロジックですね。

インタラクティブにしてみる

単純なトレーリングストップだけなら証券会社が提供している機能で実現できるので、少し手を加えてインタラクティブにしてみました。

トレーリングストップでは、以下のようなパターンがありがち。

• すこし大きなドローダウンがきて、トレンド変わってないのに決済されてしまい、利益を逃した・・
• レートが急落した時に、決済が遅れて損失が広がった・・・

これを回避できるように、Botでの強制決済に加えて、人が状況をみて決済するかどうか判断できる仕組みをいれてみます。

仕様

以下のような動作をします。

• トレーリングストップの閾値を2段階で指定できるようにして、1つ目の閾値を超えたタイミングでは警告通知を送信。

  • 通知を確認して、即時決済するか、保留するか判断できる。
  • 決済をスムーズに行えるよう、通知から1タップで決済を実行できるようにする。

• 2つ目の閾値を超えた場合、Botが建玉を決済。

  • 夜間など通知を受けとっても対処できない場合を考慮して、2つ目の閾値を超えたら、強制決済するようにしておきます。
  • なお、決済時にはOANDA JAPANから通知が送信されるので、Jijiからの通知は省略しました。

Bot(エージェント)のコード

• TrailingStopAgentが、Botの本体。これをバックテストやリアルトレードで動作させればOKです。
  • エージェントファイルの追加の方法など、Jijiの基本的な使い方はこちらをご覧ください。
• TrailingStopAgent自体は、新規に建玉を作ることはしません。
  • 裁量トレードや他のエージェントが作成した建玉を自動で監視し、トレーリングストップを行います。
  • バックテストで試す場合は、建玉を作成するエージェントと一緒に動作させてください。
• 機能の再利用ができるように、処理はTrailingStopManagerに実装しています。

# トレーリングストップで建玉を決済するエージェント
class TrailingStopAgent

  include Jiji::Model::Agents::Agent

  def self.description
    <<-STR
トレーリングストップで建玉を決済するエージェント。
 - 損益が警告を送る閾値を下回ったら、1度だけ警告をPush通知で送信。
 - さらに決済する閾値も下回ったら、建玉を決済します。
      STR
  end

  # UIから設定可能なプロパティの一覧
  def self.property_infos
    [
      Property.new('warning_limit', '警告を送る閾値', 20),
      Property.new('closing_limit', '決済する閾値',   40)
    ]
  end

  def post_create
    @manager = TrailingStopManager.new(
      @warning_limit.to_i, @closing_limit.to_i, notifier)
  end

  def next_tick(tick)
    @manager.check(broker.positions, broker.pairs)
  end

  def execute_action(action)
    @manager.process_action(action, broker.positions) || '???'
  end

  def state
    {
      trailing_stop_manager: @manager.state
    }
  end

  def restore_state(state)
    if state[:trailing_stop_manager]
      @manager.restore_state(state[:trailing_stop_manager])
    end
  end

end

# 建玉を監視し、最新のレートに基づいてトレールストップを行う
class TrailingStopManager

  # コンストラクタ
  #
  # warning_limit:: 警告を送信する閾値(pip)
  # closing_limit:: 決済を行う閾値(pip)
  # notifier:: notifier
  def initialize(warning_limit, closing_limit, notifier)
    @warning_limit = warning_limit
    @closing_limit = closing_limit
    @notifier  = notifier

    @states = {}
  end

  # 建玉がトレールストップの閾値に達していないかチェックする。
  # warning_limit を超えている場合、警告通知を送信、
  # closing_limit を超えた場合、強制的に決済する。
  #
  # positions:: 建て玉一覧(broker#positions)
  # pairs:: 通貨ペア一覧(broker#pairs)
  def check(positions, pairs)
    @states = positions.each_with_object({}) do |position, r|
      r[position.id.to_s] = check_position(position, pairs)
    end
  end

  # アクションを処理する
  #
  # action:: アクション
  # positions:: 建て玉一覧(broker#positions)
  # 戻り値:: アクションを処理できた場合、レスポンスメッセージ。
  #         TrailingStopManagerが管轄するアクションでない場合、nil
  def process_action(action, positions)
    return nil unless action =~ /trailing\_stop\_\_([a-z]+)_(.*)$/
    case $1
    when "close" then
        position = positions.find {|p| p.id.to_s == $2 }
        return nil unless position
        position.close
        return "建玉を決済しました。"
    end
  end

  # 永続化する状態。
  def state
    @states.each_with_object({}) {|s, r| r[s[0]] = s[1].state }
  end

  # 永続化された状態から、インスタンスを復元する
  def restore_state(state)
    @states = state.each_with_object({}) do |s, r|
      state = PositionState.new( nil,
        @warning_limit, @closing_limit )
      state.restore_state(s[1])
      r[s[0]] = state
    end
  end

  private

  # 建玉の状態を更新し、閾値を超えていたら対応するアクションを実行する。
  def check_position(position, pairs)
    state = get_and_update_state(position, pairs)
    if state.under_closing_limit?
      position.close
    elsif state.under_warning_limit?
      unless state.sent_warning # 通知は1度だけ送信する
        send_notification(position, state)
        state.sent_warning = true
      end
    end
    return state
  end

  def get_and_update_state(position, pairs)
    state = create_or_get_state(position, pairs)
    state.update(position)
    state
  end

  def create_or_get_state(position, pairs)
    key = position.id.to_s
    return @states[key] if @states.include? key
    PositionState.new(
      retrieve_pip_for(position.pair_name, pairs),
      @warning_limit, @closing_limit )
  end

  def retrieve_pip_for(pair_name, pairs)
    pairs.find {|p| p.name == pair_name }.pip
  end

  def send_notification(position, state)
    message = "#{create_position_description(position)}" \
      + " がトレールストップの閾値を下回りました。決済しますか?"
    @notifier.push_notification(message,  [{
        'label'  => '決済する',
        'action' => 'trailing_stop__close_' + position.id.to_s
    }])
  end

  def create_position_description(position)
    sell_or_buy = position.sell_or_buy == :sell ? "売" : "買"
    "#{position.pair_name}/#{position.entry_price}/#{sell_or_buy}"
  end

end

class PositionState

  attr_reader :max_profit, :profit_or_loss, :max_profit_time, :last_update_time
  attr_accessor :sent_warning

  def initialize(pip, warning_limit, closing_limit)
    @pip           = pip
    @warning_limit = warning_limit
    @closing_limit = closing_limit
    @sent_warning  = false
  end

  def update(position)
    @units            = position.units
    @profit_or_loss   = position.profit_or_loss
    @last_update_time = position.updated_at

    if @max_profit.nil? || position.profit_or_loss > @max_profit
      @max_profit      = position.profit_or_loss
      @max_profit_time = position.updated_at
      @sent_warning    = false
      # 高値を更新したあと、 warning_limit を超えたら再度警告を送るようにする
    end
  end

  def under_warning_limit?
    return false if @max_profit.nil?
    difference >= @warning_limit * @units * @pip
  end

  def under_closing_limit?
    return false if @max_profit.nil?
    difference >= @closing_limit * @units * @pip
  end

  def state
    {
      "max_profit"      => @max_profit,
      "max_profit_time" => @max_profit_time,
      "pip"             => @pip,
      "sent_warning"    => @sent_warning
    }
  end

  def restore_state(state)
    @max_profit      = state["max_profit"]
    @max_profit_time = state["max_profit_time"]
    @pip             = state["pip"]
    @sent_warning    = state["sent_warning"]
  end

  private

  def difference
    @max_profit - @profit_or_loss
  end

end

それでは、みなさま、良いお年を。

MongoDBのinsert/updateをまとめて、bulk insert/update に流すユーティリティを書いた

バッチ処理などでMongoDBに大量のinsert/updateを行うとき、Mongoidを使って1つずつ #save してると遅い。 ということで、複数の #save をまとめて bulk insert/update に流すユーティリティを書いてみました。

使い方

• モデルクラスで、Mongoid::Document と Utils::BulkWriteOperationSupport を include する。
• Utils::BulkWriteOperationSupport.begin_transaction を呼び出してから、モデルの #save を呼び出す。
  • この時点ではMongoDBへのinsert/updateは行われず、バッファに蓄積されます。
• Utils::BulkWriteOperationSupport.end_transaction を呼び出すと、バッファのデータが #bulk_write でまとめて永続化される。

class TestModel
  
  # Mongoid::Document と Utils::BulkWriteOperationSupport をincludeする
  include Mongoid::Document
  include Utils::BulkWriteOperationSupport

  store_in collection: 'test_model'

  field :name,      type: String

end

#略

puts TestModel.count # => 0

Utils::BulkWriteOperationSupport.begin_transaction

# #begin_transaction を呼び出したあと、モデルを作成/変更して、 #save を呼び出す。
a = TestModel.new
a.name = 'a'
b = TestModel.new
b.name = 'b'
a.save
b.save

# #end_transaction を実行するまで、永続化されない
puts TestModel.count # => 0

# #end_transaction を呼び出すと、バッファのデータが #bulk_write でまとめて永続化される。
Utils::BulkWriteOperationSupport.end_transaction

puts TestModel.count # => 2

ユーティリティのコード

• Document#save を書き換えて、#begin_transaction ～ #end_transaction の間であれば、スレッドローカルに永続化対象としてマーク。
• #end_transaction が呼び出されたタイミンクで、まとめて #bulk_write で永続化します。
• 参照整合性のチェックとか、いろいろ手抜きなので必要に応じて改造してください。

module BulkWriteOperationSupport

  KEY = BulkWriteOperationSupport.name

  def save
    if BulkWriteOperationSupport.in_transaction?
      BulkWriteOperationSupport.transaction << self
    else
      super
    end
  end

  def self.in_transaction?
    !transaction.nil?
  end

  def self.begin_transaction
    Thread.current[KEY] = Transaction.new
  end

  def self.end_transaction
    return unless in_transaction?
    transaction.execute
    Thread.current[KEY] = nil
  end

  def self.transaction
    Thread.current[KEY]
  end

  def create_insert_operation
    { :insert_one => as_document }
  end

  def create_update_operation
    {
      :update_one => {
        :filter => { :_id => id },
        :update => {'$set' => collect_changed_values }
      }
    }
  end

  private

  def collect_changed_values
    changes.each_with_object({}) do |change, r|
      r[change[0].to_sym] = change[1][1]
    end
  end

  class Transaction

    def initialize
      @targets = {}
    end

    def <<(model)
      targets_of( model.class )[model.object_id] = model
    end

    def execute
      until @targets.empty?
        model_class = @targets.keys.first
        execute_bulk_write_operations(model_class)
      end
    end

    def size
      @targets.values.reduce(0) {|a, e| a + e.length }
    end

    private

    def targets_of( model_class )
      @targets[model_class] ||= {}
    end

    def execute_bulk_write_operations(model_class)
      return unless @targets.include?(model_class)
      execute_parent_object_bulk_write_operations_if_exists(model_class)

      client = model_class.mongo_client[model_class.collection_name]
      operations = create_operations(@targets[model_class].values)
      client.bulk_write(operations) unless operations.empty?

      @targets.delete model_class
    end

    def execute_parent_object_bulk_write_operations_if_exists(model_class)
      parents = model_class.reflect_on_all_associations(:belongs_to)
      parents.each do |m|
        klass = m.klass
        execute_bulk_write_operations(klass)
      end
    end

    def create_operations(targets)
      targets.each_with_object([]) do |model, array|
        if model.new_record?
          model.new_record = false
          array << model.create_insert_operation
        else
          array << model.create_update_operation if model.changed?
        end
      end
    end

  end

end

nukeproof/oanda_api のコネクションリーク問題とその対策

OANDA fx Trade APIのRubyクライアント「nukeproof/oanda_api」には、TCPコネクションリークの問題があり、長時間連続で利用しているとファイルディスクリプタが枯渇します。

内部で利用している persistent_http の古いバージョンにある不具合が原因(最新の2.0.1では改修済み)のため、Gemfileなどで最新バージョンを使うようにすると回避できます。

gem 'persistent_http', '2.0.1'

問題の詳細

Jijiを10日程度連続稼働させていて発覚。突然、以下のエラーが発生するようになりました。

E, [2015-12-09T01:23:06.337582 #7932] ERROR -- : Too many open files - getaddrinfo (Errno::EMFILE)
/home/yamautim/.rbenv/versions/2.2.3/lib/ruby/2.2.0/net/http.rb:879:in `initialize'
/home/yamautim/.rbenv/versions/2.2.3/lib/ruby/2.2.0/net/http.rb:879:in `open'
/home/yamautim/.rbenv/versions/2.2.3/lib/ruby/2.2.0/net/http.rb:879:in `block in connect'

lsof コマンドの出力行数も少しずつ増えていきます。

$ lsof -p <Jijiのpid> | wc -l 
77
$ lsof -p <Jijiのpid> | wc -l 
79

起動直後と、しばらく起動した後で、lsofの出力結果のDiffをとると、CLOSE_WAITのhttpsコネクションが増加していました。

ruby    4389 xxxx   23u  IPv4 9265756       0t0       TCP localhost.localdomain:35773->unknown.xxxx.net:https (CLOSE_WAIT)

外向きのHTTPS通信なので、OANDA へのアクセスっぽい。

原因

最初にも書きましたが、persistent_http の古いバージョンにある不具合が原因です。最新の 2.0.1 を使えば改修されます。

• 1.0.6では、内部で利用しているGenePoolに渡すオプションがnilになっており、コネクションの破棄が正しく行われない状態になっています。
• このコミットで改修されていて、最新の 2.0.1 を使えばOK。
• ただし、oanda_api が直接依存している persistent_httparty で バージョン2以下を使うよう明示されているため、普通に使うと 1.0.6 が使われてしまいます。
• このため、Gemfileなどで最新バージョンを使うように明示する等の対応が必要です。
• ちなみに、persistent_httparty に、依存するpersistent_httpのバージョンを上げるPull Requestはあるのですが、マージされていないようです・・・。

OANDA fx Trade API を使って、リアルタイムな為替レートを取得してみる

今さらですが、OANDA Japan から FXトレードAPI が提供されているのを発見。

www.oanda.jp

• レート情報の取得から、取引、建玉情報の取得などのFX 取引に必要なAPI一式が提供されています。
• 初期費用、月額利用料金は無料。ただし、口座残高が25万円以上必要です。
• 本物の口座を使うLive環境のほか、デモ口座のアカウントで使える Practice、アカウントなしで試せるSandbox環境も提供。
• REST APIの他、Java/FIX版のAPIも有り。

機能もかなり充実してます。本格的なトレードアプリも作れるんじゃないかな。

• REST APIだけど、ストリーミングでのレートデータ配信に対応
  • Transfer-Encoding: chunked で配信する仕組みで、ポーリングなしでリアルタイムなレートデータを取得できます。
• OAuth 2.0での認証に対応
  • サードパーティのサービスに、アカウント/パスワードを渡すことなくREST APIへのアクセストークンを安全に渡すことが可能。
  • なので、クラウド型のWebサービスも作ったりしやすいかと。
  • ただし、APIから取得した情報の再配信は禁止されているのでご注意。
• すべてのGET APIが ETag に対応しているなど、細かいところの完成度も高い。

ということで、Rubyを使ってREST APIにアクセスし、リアルタイムな為替レートを取得してみるサンプルを書いてみました。(といっても、公開されているアクセスライブラリのサンプルを動かしてみただけですが・・・)

アクセスライブラリのインストール

RubyだとREST APIのアクセスライブラリが公開されているので、これを利用します。 gemでインストール。

$ gem install oanda_api

デモ口座のアカウントを使って、現在価格を得るサンプルコードです。 「<アクセストークン>」のところには、OANDAのサイトから取得したアクセストークンを指定します。取得方法はこちら。

require 'oanda_api'

client = OandaAPI::Client::TokenClient.new(:practice, <アクセストークン>)

prices = client.prices(instruments: ['EUR_USD','USD_JPY']).get
prices.each do |p|
  puts "#{p.instrument} #{p.ask} #{p.bid} #{p.time}"
end

実行結果です。

$ retrieve_price.rb
EUR_USD 1.1201 1.11994 2015-05-11 00:22:16 UTC
USD_JPY 119.744 119.728 2015-05-11 00:21:43 UTC

せっかくなので、ストリーミングAPIでリアルタイムなレート情報の取得もやってみます。

require 'oanda_api'

access_token =  <アクセストークン>

# アカウントのIDが必要なので取得しておく。
client = OandaAPI::Client::TokenClient.new(:practice, access_token )
account =  client.accounts.get.first

# ストリーミングAPI用のクライアントを別途作成し、実行
streaming_client = OandaAPI::Streaming::Client.new(:practice, access_token )
prices = streaming_client.prices(account_id: account.account_id, instruments: ["USD_JPY"])
prices.stream do |p|
  # ※無限ループするので Ctrl+Cで停止すること。
  puts "#{p.instrument} #{p.ask} #{p.bid} #{p.time}"
end 

実行してみます。放置するとずっと更新を続けるので、Ctrl+Cで停止すること。

$ ruby retrieve_price_uses_streaming_api.rb
USD_JPY 119.747 119.743 2015-05-11 00:37:52 UTC
USD_JPY 119.748 119.744 2015-05-11 00:38:12 UTC
USD_JPY 119.749 119.745 2015-05-11 00:38:12 UTC
USD_JPY 119.75 119.746 2015-05-11 00:38:12 UTC
...

Jiji2のバックエンドもこれに移行かな

Jiji2絶賛開発中ですが、

unageanu.hatenablog.com

github.com

バックエンドは OANDA fx trade API に変更かな、と考えています。 旧バージョンと同じくスクレイピングを使う方向で考えていましたが、次のようなリスクはどうしても残るので。

• (アクセス頻度はもちろん抑えるにしても)やはり証券会社のサービスに多少は負荷がかかるので、利用禁止になる可能性がある。
• サイトの仕様変更で突然動作しなくなるリスクがある。

アプリ開発者としては、特定の証券会社にロックオンしてしまうのは避けたいところではあるので、このようなAPIを提供してくれる証券会社がもっと増えてくるといいな。(クリック証券、何でやめたんや・・・。)

追記(2015-12-01)

OANDA FX trade APIを利用した、無料のFXシステムトレードフレームワーク「Jiji」をリリースしました!

jiji2.unageanu.net

使ってみて、ご意見など頂けるとうれしいです。

設定値管理ライブラリ Figaroを使う

Figaro は Rubyアプリ用のシンプルな設定値管理ライブラリ です。

github.com

• 設定ファイル(YAML)の値を読み込んで、アプリから参照できるようにします。
  • 設定ファイルは1つ。その中に、環境(development,production,test)ごとの設定も書く形。
• アクセス時のインターフェイスに ENV を使うのが特徴。
  • ENV を 拡張し、設定ファイルから読み込んだ値も ENV['HOGE'] で参照できるようにしてくれます。
  • なので、環境変数で設定値を管理するやり方と親和性が高い。
    Heroku環境では、このやり方が主流になっているようです。

同様の機能を提供するライブラリに、 Dotenv もあります。違いはざっくり以下の3つ。

違い	Figaro	Dotenv
設定ファイルの形式	YAML	独自のKey-Value形式
環境ごとの設定の切り分け	1つのYAMLの中に書く形	ファイルを別ける
Rails	Railsでの利用にフォーカス	非Rails環境もサポート

設定ファイルはYAMLのほうがいいかな・・・、と思ってFigaroにしました。参照使えるしね。ただし、Rails向けに作られているところがあるので、非Rails環境で使うには一工夫必要です。

使い方

インストール:

$ gem install figaro

設定ファイル( ./config/application.yml )を用意ます。

# デフォルトの設定
KEY1: 'key1'
KEY2: 'key2-default'
# ERBで処理されるので、rubyコードも利用可
ERB : <%= 'abc' * 3 %>

# 特定環境での設定
test:
  KEY2: 'key2-test'

development:
  KEY2: 'key2-development'

• ルートに、キーと値を書きます。
  • アクセス時のインターフェイスはENVなので、階層を作ったりはできません。
• test や development のようなエントリーを用意して、特定環境下でのみ有効な設定を記載できます。
  • 値の優先順位は、以下の通りです。
    • 1.環境変数の値
    • 2.特定環境下向けの設定
    • 3.デフォルトの設定
• ERBで処理されるので、rubyコードも書けます。

設定ファイルを使うサンプルです。

require 'figaro'

class Application < Figaro::Application
  private

  # 設定ファイルのパス。 ./config/application.yml から読み込む
  def default_path
    File.join( File.dirname(__FILE__), 'config', 'application.yml')
  end
  # 環境(development,test..)の取得先。環境変数RACK_ENVを使う
  def default_environment
    ENV['RACK_ENV']
  end
end

Figaro.adapter = Application
Figaro.load

puts ENV['KEY1']
puts ENV['KEY2']
puts ENV['ERB_VALUE']

• Figaro::Application を継承した Application を用意して、設定ファイルのパスと環境の取得先を指定します。
• あとは、 Figaro.load すれば ENV で値を参照できるようになります。

実行してみます。 まずは素で実行。デフォルトの設定が使われます。

$ ruby test.rb 
key1
key2-default
abcabcabc

RACK_ENVを指定すると、特定環境用の設定がアクティブになります。

$ RACK_ENV=test ruby test.rb                                                                                                       
key1
key2-test
abcabcabc
$ RACK_ENV=development ruby test.rb 
key1
key2-development
abcabcabc

さらに、環境変数で値を上書きすることも可能です。

$ KEY1=foo KEY2=var RACK_ENV=test ruby test.rb                                                                                     
WARNING: Skipping key "KEY1". Already set in ENV.
WARNING: Skipping key "KEY2". Already set in ENV.
foo
var
abcabcabc

警告出ますけど。