如何实现trading.ipynb运行时，在monitorTrades.ipynb中执行checkFunc()并传参？

下面是几种可行的实现方式，可根据场景选择：

方式一：文件传递参数 + 本地监控 #

适合轻量、低实时性需求的场景：

• 在trading.ipynb中，计算出需要传递的2个参数后，写入临时JSON文件：

  import json
  
  # trading中计算得到的参数示例
  trade_param1 = 150
  trade_param2 = "EURUSD"
  
  # 写入临时文件
  with open("temp_trade_params.json", "w") as f:
      json.dump({"tp1": trade_param1, "tp2": trade_param2}, f)
  

• 在monitorTrades.ipynb中，循环监控文件变化，读取参数并执行checkFunc()：

  import json
  import time
  from os.path import getmtime
  
  # monitor内部的参数示例
  monitor_param1 = 0.02
  monitor_param2 = True
  
  def checkFunc(p1, p2, p3, p4):
      # 自定义函数逻辑
      print(f"执行checkFunc：{p1}, {p2}, {p3}, {p4}")
  
  last_modify_time = 0
  while True:
      try:
          current_mtime = getmtime("temp_trade_params.json")
          if current_mtime > last_modify_time:
              last_modify_time = current_mtime
              # 读取trading传入的参数
              with open("temp_trade_params.json", "r") as f:
                  trade_params = json.load(f)
              # 在monitor环境内执行函数
              checkFunc(trade_params["tp1"], trade_params["tp2"], monitor_param1, monitor_param2)
          time.sleep(1)  # 每秒检查一次文件更新
      except KeyboardInterrupt:
          break
  

方式二：Redis消息队列传递参数 #

适合高实时性需求的场景，需提前安装redis库：

• 在trading.ipynb中，将参数发送到Redis频道：

  import redis
  import json
  
  r = redis.Redis(host='localhost', port=6379, db=0)
  trade_param1 = 150
  trade_param2 = "EURUSD"
  
  # 序列化并发送参数
  r.publish("trade_signal_channel", json.dumps({"tp1": trade_param1, "tp2": trade_param2}))
  

• 在monitorTrades.ipynb中，订阅频道接收参数并执行函数：

  import redis
  import json
  
  monitor_param1 = 0.02
  monitor_param2 = True
  
  def checkFunc(p1, p2, p3, p4):
      print(f"执行checkFunc：{p1}, {p2}, {p3}, {p4}")
  
  r = redis.Redis(host='localhost', port=6379, db=0)
  pubsub = r.pubsub()
  pubsub.subscribe("trade_signal_channel")
  
  # 持续监听频道消息
  for msg in pubsub.listen():
      if msg['type'] == 'message':
          trade_params = json.loads(msg['data'].decode('utf-8'))
          checkFunc(trade_params["tp1"], trade_params["tp2"], monitor_param1, monitor_param2)
  

方式三：HTTP服务触发执行 #

适合跨环境调用的场景，需提前安装flask库：

1. 新建common_utils.py文件，将checkFunc()提取为公共模块：

   def checkFunc(p1, p2, p3, p4):
       print(f"执行checkFunc：{p1}, {p2}, {p3}, {p4}")
   

2. 在monitorTrades.ipynb中启动HTTP服务，接收参数并执行函数：

   from flask import Flask, request
   from common_utils import checkFunc
   
   monitor_param1 = 0.02
   monitor_param2 = True
   
   app = Flask(__name__)
   
   @app.route('/trigger-check', methods=['POST'])
   def trigger_check():
       req_data = request.json
       checkFunc(req_data["tp1"], req_data["tp2"], monitor_param1, monitor_param2)
       return {"status": "executed"}
   
   if __name__ == '__main__':
       app.run(host='localhost', port=5000)
   

3. 在trading.ipynb中发送POST请求，触发monitor端的函数执行：

   import requests
   
   trade_param1 = 150
   trade_param2 = "EURUSD"
   
   requests.post("http://localhost:5000/trigger-check", json={"tp1": trade_param1, "tp2": trade_param2})
   

内容的提问来源于stack exchange，提问作者Sanchay Kasturey