def initialize(self):
  """在回測開始前的初始化"""
  pass
 def before_on_tick(self, tick):
  pass
 def after_on_tick(self, tick):
  pass
 def before_trade(self, order):
  """在交易之前會調(diào)用此函數(shù)
  可以在此放置資金管理及風(fēng)險管理的代碼
  如果返回True就允許交易，否則放棄交易
  """
  return True
 def on_order_ok(self, order):
  """當訂單執(zhí)行成功后調(diào)用"""
  pass
 def on_order_timeout(self, order):
  """當訂單超時后調(diào)用"""
  pass
 def finish(self):
  """在回測結(jié)束后調(diào)用"""
  pass
 @abstractmethod
 def on_tick(self, bar):
  """
  回測實例必須實現(xiàn)的方法，并編寫自己的交易邏輯
  """
  pass

def buy(self, code, price, shares, ttl=-1):
  """
  限價提交買入訂單
  ---------
  Parameters:
   code:str
    股票代碼
   price:float or None
    最高可買入的價格, 如果為None則按市價買入
   shares:int
    買入股票數(shù)量
   ttl:int
    訂單允許存在的最大時間，默認為-1，永不超時
  ---------
  return:
   dict
    {
    "type": 訂單類型, "buy",
    "code": 股票代碼,
    "date": 提交日期,
    "ttl": 存活時間, 當ttl等于0時則超時，往后不會在執(zhí)行
    "shares": 目標股份數(shù)量,
    "price": 目標價格,
    "deal_lst": 交易成功的歷史數(shù)據(jù)，如
     [{"price": 成交價格,
      "date": 成交時間,
      "commission": 交易手續(xù)費,
      "shares": 成交份額
     }]
    ""
   }
  """
  if price is None:
   stock_info = self.ctx.tick_data[code]
   price = stock_info[self.deal_price]
  order = {
   "type": "buy",
   "code": code,
   "date": self.ctx.now,
   "ttl": ttl,
   "shares": shares,
   "price": price,
   "deal_lst": []
  }
  self.submit(order)
  return order
 def sell(self, code, price, shares, ttl=-1):
  """
  限價提交賣出訂單
  ---------
  Parameters:
   code:str
    股票代碼
   price:float or None
    最低可賣出的價格, 如果為None則按市價賣出
   shares:int
    賣出股票數(shù)量
   ttl:int
    訂單允許存在的最大時間，默認為-1，永不超時
  ---------
  return:
   dict
    {
    "type": 訂單類型, "sell",
    "code": 股票代碼,
    "date": 提交日期,
    "ttl": 存活時間, 當ttl等于0時則超時，往后不會在執(zhí)行
    "shares": 目標股份數(shù)量,
    "price": 目標價格,
    "deal_lst": 交易成功的歷史數(shù)據(jù)，如
     [{"open_price": 開倉價格,
      "close_price": 成交價格,
      "close_date": 成交時間,
      "open_date": 持倉時間,
      "commission": 交易手續(xù)費,
      "shares": 成交份額,
      "profit": 交易收益}]
    ""
   }
  """
  if code not in self.position:
   return
  if price is None:
   stock_info = self.ctx.tick_data[code]
   price = stock_info[self.deal_price]
  order = {
   "type": "sell",
   "code": code,
   "date": self.ctx.now,
   "ttl": ttl,
   "shares": shares,
   "price": price,
   "deal_lst": []
  }
  self.submit(order)
  return order

class Scheduler(object):
 """

 ctx.get_hist
 """
 def __init__(self):
  """"""
  self.ctx = Context()
  self._pre_hook_lst = []
  self._post_hook_lst = []
  self._runner_lst = []
 def run(self):
  # runner指存在可調(diào)用的initialize, finish, run(tick)的對象
  runner_lst = list(chain(self._pre_hook_lst, self._runner_lst, self._post_hook_lst))
  # 循環(huán)開始前為broker, backtest, hook等實例綁定ctx對象及調(diào)用其initialize方法
  for runner in runner_lst:
   runner.ctx = self.ctx
   runner.initialize()
  # 創(chuàng)建交易日歷
  if "trade_cal" not in self.ctx:
   df = list(self.ctx.feed.values())[0]
   self.ctx["trade_cal"] = df.index
  # 通過遍歷交易日歷的時間依次調(diào)用runner
  # 首先調(diào)用所有pre-hook的run方法
  # 然后調(diào)用broker,backtest的run方法
  # 最后調(diào)用post-hook的run方法
  for tick in self.ctx.trade_cal:
   self.ctx.set_currnet_time(tick)
   for runner in runner_lst:
    runner.run(tick)
  # 循環(huán)結(jié)束后調(diào)用所有runner對象的finish方法
  for runner in runner_lst:
   runner.finish()

class Context(UserDict):
 def __getattr__(self, key):
  # 讓調(diào)用這可以通過索引或者屬性引用皆可
  return self[key]
 def set_currnet_time(self, tick):
  self["now"] = tick
  tick_data = {}
  # 獲取當前所有有報價的股票報價
  for code, hist in self["feed"].items():
   df = hist[hist.index == tick]
   if len(df) == 1:
    tick_data[code] = df.iloc[-1]
  self["tick_data"] = tick_data
 def get_hist(self, code=None):
  """如果不指定code, 獲取截至到當前時間的所有股票的歷史數(shù)據(jù)"""
  if code is None:
   hist = {}
   for code, hist in self["feed"].items():
    hist[code] = hist[hist.index <= self.now]
  elif code in self.feed:
   return {code: self.feed[code]}
  return hist
class Scheduler(object):
 """

  ctx.feed: {code1: pd.DataFrame, code2: pd.DataFrame}對象
  ctx.now: 循環(huán)所處時間
  ctx.tick_data: 循環(huán)所處時間的所有有報價的股票報價
  ctx.trade_cal: 交易日歷
  ctx.broker: Broker對象
  ctx.bt/ctx.backtest: Backtest對象

  ctx.get_hist
 """
 def __init__(self):
  """"""
  self.ctx = Context()
  self._pre_hook_lst = []
  self._post_hook_lst = []
  self._runner_lst = []
 def add_feed(self, feed):
  self.ctx["feed"] = feed
 def add_hook(self, hook, typ="post"):
  if typ == "post" and hook not in self._post_hook_lst:
   self._post_hook_lst.append(hook)
  elif typ == "pre" and hook not in self._pre_hook_lst:
   self._pre_hook_lst.append(hook)
 def add_broker(self, broker):
  self.ctx["broker"] = broker
 def add_backtest(self, backtest):
  self.ctx["backtest"] = backtest
  # 簡寫
  self.ctx["bt"] = backtest
 def add_runner(self, runner):
  if runner in self._runner_lst:
   return
  self._runner_lst.append(runner)

class Stat(Base):
  def __init__(self):
    self._date_hist = []
    self._cash_hist = []
    self._stk_val_hist = []
    self._ast_val_hist = []
    self._returns_hist = []

  def run(self, tick):
    self._date_hist.append(tick)
    self._cash_hist.append(self.ctx.broker.cash)
    self._stk_val_hist.append(self.ctx.broker.stock_value)
    self._ast_val_hist.append(self.ctx.broker.assets_value)

  @property
  def data(self):
    df = pd.DataFrame({"cash": self._cash_hist,
              "stock_value": self._stk_val_hist,
              "assets_value": self._ast_val_hist}, index=self._date_hist)
    df.index.name = "date"
    return df

def get_dropdown(self):
    high_val = -1
    low_val = None
    high_index = 0
    low_index = 0
    dropdown_lst = []
    dropdown_index_lst = []
    for idx, val in enumerate(self._ast_val_hist):
      if val >= high_val:
        if high_val == low_val or high_index >= low_index:
          high_val = low_val = val
          high_index = low_index = idx
          continue
        dropdown = (high_val - low_val) / high_val
        dropdown_lst.append(dropdown)
        dropdown_index_lst.append((high_index, low_index))
        high_val = low_val = val
        high_index = low_index = idx
      if low_val is None:
        low_val = val
        low_index = idx
      if val < low_val:
        low_val = val
        low_index = idx
    if low_index > high_index:
      dropdown = (high_val - low_val) / high_val
      dropdown_lst.append(dropdown)
      dropdown_index_lst.append((high_index, low_index))
    return dropdown_lst, dropdown_index_lst
  @property
  def max_dropdown(self):
    """最大回車率"""
    dropdown_lst, dropdown_index_lst = self.get_dropdown()
    if len(dropdown_lst) > 0:
      return max(dropdown_lst)
    else:
      return 0
  @property
  def annual_return(self):
    """
    年化收益率
    y = (v/c)^(D/T) - 1
    v: 最終價值
    c: 初始價值
    D: 有效投資時間(365)

    """
    D = 365
    c = self._ast_val_hist[0]
    v = self._ast_val_hist[-1]
    days = (self._date_hist[-1] - self._date_hist[0]).days
    ret = (v / c) ** (D / days) - 1
    return ret

import json
from backtest import BackTest
from reporter import Plotter
class MyBackTest(BackTest):
  def initialize(self):
    self.info("initialize")
  def finish(self):
    self.info("finish")
  def on_tick(self, tick):
    tick_data = self.ctx["tick_data"]
    for code, hist in tick_data.items():
      if hist["ma10"] > 1.05 * hist["ma20"]:
        self.ctx.broker.buy(code, hist.close, 500, ttl=5)
      if hist["ma10"] < hist["ma20"] and code in self.ctx.broker.position:
        self.ctx.broker.sell(code, hist.close, 200, ttl=1)
if __name__ == '__main__':
  from utils import load_hist
  feed = {}
  for code, hist in load_hist("000002.SZ"):
    # hist = hist.iloc[:100]
    hist["ma10"] = hist.close.rolling(10).mean()
    hist["ma20"] = hist.close.rolling(20).mean()
    feed[code] = hist
  mytest = MyBackTest(feed)
  mytest.start()
  order_lst = mytest.ctx.broker.order_hist_lst
  with open("report/order_hist.json", "w") as wf:
    json.dump(order_lst, wf, indent=4, default=str)
  stats = mytest.stat
  stats.data.to_csv("report/stat.csv")
  print("策略收益： {:.3f}%".format(stats.total_returns * 100))
  print("最大回徹率: {:.3f}% ".format(stats.max_dropdown * 100))
  print("年化收益: {:.3f}% ".format(stats.annual_return * 100))
  print("夏普比率: {:.3f} ".format(stats.sharpe))
  plotter = Plotter(feed, stats, order_lst)
  plotter.report("report/report.png")