moomoo証券の注文履歴CSVから実現損益を算出する

moomoo証券のUIは一見モダンだが、実績の確認が死ぬほどしづらく、資産残高が正しいかどうかの検証が非常に面倒くさい。 そのため、pythonを用いて、注文履歴のCSVから実現損益を計算するスクリプトを作った。

1. 適当な場所から、usdjpyの日足終値のデータを見つけて、usdjpy.csvとして保存
2. moomoo証券の口座→CashまたはMargin→History→右上のハンバーガーメニュー→Exportから、CSVを保存
3. 下のスクリプトにcsvを食わせる

左から順に、日付、費用(手数料+損失)、収益、利益の順に表示される。

$ python3 analyze.py cash cash.csv | column -t
12/30/2024  0       9952   9952
12/31/2024  0       7226   7226
01/06/2025  0       37993  37993
01/07/2025  0       23524  23524
01/10/2025  5478    0      -5478
01/14/2025  24807   0      -24807
01/30/2025  0       10974  10974
01/31/2025  0       12509  12509

$ python3 analyze.py margin margin.csv | column -t
01/06/2025  1598    0      -1598
01/15/2025  1458    41836  40378
01/16/2025  41590   10245  -31345
01/17/2025  36996   26189  -10807
01/21/2025  11257   32657  21400
01/23/2025  28821   12016  -16805
01/24/2025  10897   34714  23817
01/27/2025  35918   21044  -14874
01/28/2025  16843   23690  6847
01/29/2025  111954  25349  -86605
01/30/2025  30661   25189  -5472
01/31/2025  20618   47454  26836
02/02/2025  751     0      -751
02/03/2025  3884    50231  46347

import csv
import sys
from datetime import datetime
 
mode = sys.argv[1]
file = sys.argv[2]
 
# ---- usdjpyデータ読み込み
usdjpy = dict()
 
with open("usdjpy.csv", "r", encoding="utf-8") as f:
  for idx, row in enumerate(csv.reader(f)):
    if idx == 0: continue
    usdjpy[row[0]] = float(row[1])
 
expense = dict()
revenue = dict()
days    = list()
 
# ---- modeに応じてCSVのカラムインデックスを導く
if mode == "margin":
  side_idx   = 0
  sym_idx    = 1
  qty_idx    = 20
  amount_idx = 22
  dt_idx     = 23
  fee_idx    = 33
 
  fee_ratio = 0.00132
 
  is_open  = lambda x: x.endswith("Open")
  is_close = lambda x: x.endswith("Close")
  is_long  = lambda x: x.startswith("Buy") == x.endswith("Open")
  is_short = lambda x: x.startswith("Sell") == x.endswith("Open")
elif mode == "cash":
  side_idx   = 0
  sym_idx    = 1
  qty_idx    = 18
  amount_idx = 20
  dt_idx     = 21
  fee_idx    = 30
 
  fee_ratio = 0
 
  is_open  = lambda x: x == "Buy"
  is_close = lambda x: x == "Sell"
  is_long  = lambda x: True
  is_short = lambda x: False
else:
  assert False
 
# ---- CSV 読み込み
with open(file, "r", encoding="utf-8") as f:
  rows = []
  for idx, row in enumerate(csv.reader(f)):
    if idx == 0: continue
 
    if row[qty_idx] == "": continue
 
    side   = row[side_idx]
    sym    = row[sym_idx]
    qty    = float(row[qty_idx])
    amount = float(row[amount_idx].replace(",", ""))
    dt     = datetime.strptime(row[dt_idx], "%b %d, %Y %H:%M:%S ET")
    if len(row) > fee_idx:
      fee = float(row[fee_idx].replace(",", "")) if row[fee_idx] != "" else 0
    else:
      fee = min(fee_ratio * amount, 22)
 
    if side == "":
      pr       = rows[-1]
      rows[-1] = (pr[0], pr[1], pr[2], pr[3]+qty, pr[4]+amount, pr[5], pr[6]+fee)
    else:
      if is_long(side):
        type_ = "Long"
      elif is_short(side):
        type_ = "Short"
      else:
        assert False
      rows.append((side, type_, sym, qty, amount, dt, fee))
 
  x = 0
  positions = {}
  for row in rows[::-1]:
    side, type_, sym, qty, amount, dt, fee = row
 
    dstr = dt.date().strftime("%m/%d/%Y")
    if dstr not in days:
      days.append(dstr)
 
    p = f"{sym}/{type_}"
    s = positions.get(p, (0, 0))
    if is_open(side):
      positions[p] = (s[0]+qty, s[1]+amount)
      expense[dstr] = expense.get(dstr, 0) + fee
    elif is_close(side):
      avg = s[1]/s[0]
      positions[p] = (s[0]-qty, s[1]-avg*qty)
 
      if type_ == "Long":
        earn = amount - avg*qty
      elif type_ == "Short":
        earn = avg*qty - amount
      else:
        assert False
 
      expense[dstr] = expense.get(dstr, 0) + fee
      if earn > 0:
        revenue[dstr] = revenue.get(dstr, 0) + earn
      else:
        expense[dstr] = expense.get(dstr, 0) - earn
 
for day in days:
  d = usdjpy[day]
  e = int(expense.get(day, 0)*d)
  r = int(revenue.get(day, 0)*d)
  print(day, e, r, r-e)