1、vnpy系统缺少音乐和语音播放功能
vnpy中多采用各种应用系统的策略进行交易的，虽然也有各种日志和提示出现，但平常总是静悄悄的。
如果你想了解系统和策略的运行情况，可以查看各种运行日志，例如MainWindow的日志，委托列表，成交的列表，账户列表。想要查询你的策略运行情况，可以查看你的策略管理器的变量输出，等等。可是人总不能一直盯着屏幕，那样太累了。
如果能够有个声音和语音播报各种交易活动，用户会及时得到提醒。例如：
- 当网络连接和断开时可以提醒特定的音乐或语音，可以让你及时处理网络故障；
- 用户策略中可以添加音乐或语音，可以提醒策略的交易所发生的交易活动是否正常；
- 当你的账户资金出现保证金不足，可以提醒及时入金；
- 如果逆勢日内交易，你可以设置提前3分钟提醒你即将收盘了。
2、实现方法
2.1 音乐和语音播放器的实现
在vnpy\usertools目录下添加文件sound_player.py，内容如下：
"""
多线程音乐和文本播放器,介绍如下：
特点：
既可以播放wav,mp3 等格式，有可以对文本进行播放。
使用消息引擎创建该音乐播放器,多线程并行播放声音,不会因为播放声音而阻塞业务流程。
假设其他应用或者策略中需要播放声音, sound_name为字符型的声音文件名称，使用方法有两种:
方法1: 先获取消息引擎event_engine(注：与SoundPlayer实例成交时使用的消息引擎是
相同的)，那么可以这样播放：
event_engine.put(Event(EVENT_SOUND,"Connected.wav"))
event_engine.put(Event(EVENT_SPEEK,"您收到一条委托单"))
方法2: 将SoundPlayer多play_sound()接口安装到MainEngine到实例main_engine，那么
可以先回去获取main_engine,然后这样播放：
event_engine.play_sound("Connected.wav")
event_engine.speek_text("您收到一条委托单")
作者:hxxjava 时间:2023-2-14,情人节————献给心爱的人！
修改:增加回测与实盘的区分功能，使得只在实盘环境才播放声音和文本。
修改:hxxjava 时间:2023-2-28
依赖库：pyttsx3, 安装：pip install pyttsx3
"""
from typing import Any
from pathlib import Path
from threading import Thread
from vnpy.trader.engine import EventEngine,Event
from winsound import PlaySound,SND_FILENAME
import pyttsx3
EVENT_SOUND = "eSound."
EVENT_SPEEK = "eSpeak."
class SoundPlayer():
"""
多线程声音播放器
"""
def __new__(cls, *args, **kwargs):
""" singleton constructor """
if not hasattr(cls, "_instance"):
cls._instance = super(SoundPlayer, cls).__new__(cls)
return cls._instance
def __init__(self,event_engine:EventEngine,switch:bool=True):
""" 初始化函数 """
self.event_engine = event_engine
# control play sound file
self.switch = switch
self.register_event()
def register_event(self):
""" """
self.event_engine.register(EVENT_SOUND,self.process_sound_event)
self.event_engine.register(EVENT_SPEEK,self.process_speak_event)
def set_switch(self,switch:bool=True):
""" set the swith which control play sound file """
self.switch = switch
def _get_sound_path(self,sound_name: str):
"""
Get path for sound file with sound name.
"""
this_file_path:Path = Path(__file__).parent
sound_path:Path = this_file_path.joinpath("sounds", sound_name)
return str(sound_path)
def process_sound_event(self,event:Event):
""" EVENT消息处理过程 """
wavname,is_testing = event.data['wavname'],event.data['is_testing']
if self.switch == True and is_testing == False:
filename = self._get_sound_path(wavname)
thread = Thread(target=self._play_sound,kwargs=({"filename":filename}),daemon=True)
thread.start()
def process_speak_event(self,event:Event):
""" EVENT消息处理过程 """
santence,is_testing = event.data['santence'],event.data['is_testing']
if self.switch == True and is_testing == False:
santence:str = event.data
thread = Thread(target=self._do_speak,kwargs=({"santence":santence}),daemon=True)
thread.start()
def _play_sound(self,filename:str):
""" 音乐文件播放线程执行过程 """
PlaySound(filename,SND_FILENAME)
def _do_speak(self,santence:str):
""" 文本播放线程执行过程 """
print(santence)
speaker = pyttsx3.init()
speaker.say(santence)
speaker.runAndWait()
def play_sound(self,sound_name:str,is_testing:bool=False):
"""
用户音乐播放接口。
参数：
sound_name:传入声音文件名
is_testing:回测=True;实盘=False(默认)
"""
self.event_engine.put(Event(EVENT_SOUND,{"wavname":sound_name,"is_testing":is_testing}))
def speak_text(self,santence:str,is_testing:bool=False):
"""
用户文字播放接口。
参数：
santence:传入声音文件名
is_testing:回测=True;实盘=False(默认)
"""
self.event_engine.put(Event(EVENT_SPEEK,{"santence":santence,"is_testing":is_testing}))2.2 把音乐和语音播放器安装到vnpy系统
在vnpy\trader\engine.py中做如下修改：
1）在引用部分添加这些内容
from vnpy.usertools.sound_player import SoundPlayer2）在class OmsEngine的init()中创建音乐和语音播放器
self.sound_player = SoundPlayer(event_engine,True) # test sound player3）在class OmsEngine的add_function()函数中为MainEngine添加下面的函数
def add_function(self) -> None:
"""Add query function to main engine."""
... ...
self.main_engine.play_sound = self.sound_player.play_sound
self.main_engine.speak_text = self.sound_player.speak_text这样你的MainEngine就有了可以音乐和语音功能了。
2.3 音频文件存放在哪里？
class sound_player规定音频文件存放在vnpy\usertools\sounds\目录下，当然你也可以修改代码中规定的目录，放在自己喜欢的目录下。
文件可以是wav、mp3格式的音乐文件均可，可以自己录制。
取一些有意义的文件名，如connected.wav代表网络连接成功，disconnection.wav代表网络断开，自己发挥吧，方便自己在自己vnpy系统中用函数调用。
本来本人有一套音乐文件的，可是论坛里没有文件上传功能，所以无法共享给大家，如果需要可以私信我。
3. 如何使用音乐和语音播放功能
下面用连接网关成功和连接断开，分别给出音乐和语音播放的示例：
3.1 音乐播放功能使用
3.1.1 通过main_engine调用play_sound()播放语音
def process_connect_event(self, event: Event) -> None: # hxxjava add
""" CTP接口连接消息处理 """
gateway:GatewayData = event.data
self.main_engine.play_sound("Connected.wav")
def process_disconnect_event(self, event: Event) -> None: # hxxjava add
""" CTP接口断开消息处理 """
gateway:GatewayData = event.data
self.main_engine.play_sound("ConnectionLost.wav")3.1.2 发送EVENT_SOUND消息播放音乐
# 增加引用
from vnpy.usertools.sound_player import EVENT_SOUND,EVENT_SPEEK
def on_order(self, order: OrderData):
"""
Callback of new order data update.
"""
# 当策略收到委托单时播放提示音乐
event_engine:EventEngine = self.cta_engine.event_engine
is_testing = self.cta_engine.get_engine_type() != EngineType.LIVE
event = Event(EVENT_SOUND,
{"wavname":"order.wav","is_testing":is_testing}
)
event_engine.put(event)
def on_trade(self, trade: TradeData):
"""
Callback of new trade data update.
"""
# 当策略收到成交单时播放提示音乐
event_engine:EventEngine = self.cta_engine.event_engine
is_testing = self.cta_engine.get_engine_type() != EngineType.LIVE
event = Event(EVENT_SOUND, {"wavname":"traded.wav","is_testing":is_testing} )
event_engine.put(event)3.2 语音播放功能使用
3.2.1 通过main_engine调用speak_text()播放语音
def process_connect_event(self, event: Event) -> None: # hxxjava add
""" CTP接口连接消息处理 """
gateway:GatewayData = event.data
self.main_engine.speak_text(f"感谢您，连接{gateway.name}的{gateway.type}接口成功！")
def process_disconnect_event(self, event: Event) -> None: # hxxjava add
""" CTP接口断开消息处理 """
gateway:GatewayData = event.data
self.main_engine.speak_text(f"请注意：{gateway.name}的{gateway.type}接口已断开！")3.2.2 发送EVENT_SPEEK消息播放语音
假设你的策略中实现了on_order()和on_trade()这两个回调函数：
def on_order(self, order: OrderData):
"""
Callback of new order data update.
"""
event_engine:EventEngine = self.cta_engine.event_engine
is_testing = self.cta_engine.get_engine_type() != EngineType.LIVE
event = Event(EVENT_SPEEK,
{"santance":f"策略{self.strategy_name}收到委托单，价格{order.price}，手数{order.volume}，已经成交{order.traded}",
"is_testing":is_testing})
event_engine.put(event)
def on_trade(self, trade: TradeData):
"""
Callback of new trade data update.
"""
event_engine:EventEngine = self.cta_engine.event_engine
is_testing = self.cta_engine.get_engine_type() != EngineType.LIVE
event = Event(EVENT_SPEEK,
{"santance":f"策略{self.strategy_name}收到{trade.vt_symbol}成交单，成交价{trade.price}手数{trade.volume}.",
"is_testing":is_testing})
event_engine.put(event)3.3 音乐和语音播放功能使用注意事项
实盘中用户策略是可以通过应用应用引擎获得vnpy系统的MainEngine的，这样就可以使用 play_sound() 和 speak_text()函数来播放音乐和语音了。但是，在策略中使用这个两个播放函数，应该考虑到回测时不要有声音的。应该根据应用引擎的不同，在策略中使用 play_sound() 和 speak_text()时，将参数is_testing设置为True，这样策略回测就不会有音乐和语音了。
4 将音乐和语音播放功能封装到CTA策略中
这里以CTA策略模板CtaTemplate为例，演示如何将音乐和语音播放功能封装到各种应用的策略中，其他应用系统的模板可以参考以下的做法去封装，就不再一一讲解。
# 在引用部分增加对音乐和语音播器的引用
from vnpy.usertools.sound_player import SoundPlayer
class CtaTemplate(ABC):
""""""
author: str = ""
parameters: list = []
variables: list = []
def __init__(
self,
cta_engine: Any,
strategy_name: str,
vt_symbol: str,
setting: dict,
) -> None:
""""""
... ... # 原来的初始化代码
# 音乐和语音播放器 hxxjava add
self.sound_player:SoundPlayer = SoundPlayer(self.cta_engine.event_engine)
def play_sound(self,sound_name:str):
""" 播放音乐 hxxjava add """
if self.cta_engine.get_engine_type() == EngineType.LIVE:
self.sound_player.play_sound(sound_name)
def speak_text(self,santence:str):
""" 播放语音 hxxjava add """
if self.cta_engine.get_engine_type() == EngineType.LIVE:
self.sound_player.speak_text(santence)经过上面对CtaTemplate的修改，用户策略中就可以像下面的语句一样直接调用音乐和语音播放了，更加简便。
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化开始")
self.load_bar(10)
self.speak_text(f"策略{self.strategy_name}开始初始化")
