使用 Websocket API 向 Refinitiv Real-Time 发布数据

    	
            wsa = connect("myADS",15000, "umer.nalla")
        
        Copy
    

    	
            
def connect(hostname, port, username, pos = socket.gethostbyname(socket.gethostname()), appid = 256):
    
    """ Called to connect to server """
    
    global app_id, user, position,web_socket_app
    # appid may be allocated by your Market Data team, otherwise default to 256
    app_id = appid
    user = username
    position = pos
    # the above values are stored for later usage when we attempt login.
 
    # Start websocket handshake
    ws_address = "ws://{}:{}/WebSocket".format(hostname, port)
    print("Connecting to WebSocket " + ws_address + " ...")
    web_socket_app = websocket.WebSocketApp(ws_address, header=['User-Agent: Python'],
                                        on_message=on_message,
                                        on_error=on_error,
                                        on_close=on_close,
                                        subprotocols=['tr_json2'])
    
    # callback for once websocket is open - which will send the Login request
    web_socket_app.on_open = on_open
 
    # Create Thread for WebsocketApp processing
    wst = threading.Thread(target=web_socket_app.run_forever)
    wst.start()

        
        Copy
    

    	
            
print ("Waiting for Login response")
 while not logged_in and (not shutdown_app):
     time.sleep(1)

        
        Copy
    

    	
            
def on_open(ws):
    """ Called when handshake is complete and websocket is open. Now send login """
    global web_socket_open
    web_socket_open = True
    send_login_request(ws)

        
        Copy
    

    	
            
def send_login_request(ws, is_refresh_token=False):
    """ Generate a login request and send """
    # Set values for ADS login
    # Note StreamID is 1 and the Domain is Login
    login_json = {
        'ID': 1,
        'Domain': 'Login',
        'Key': {
            'Name': '',
            'Elements': {
                'ApplicationId': '',
                'Position': ''
            }
        }
    }
 
    login_json['Key']['Name'] = user
    login_json['Key']['Elements']['ApplicationId'] = app_id
    login_json['Key']['Elements']['Position'] = position
    
     ws.send(json.dumps(login_json))

        
        Copy
    

    	
            
{
"Domain":"Login",
  "ID":1,
  "Key":{
    "Elements":{
      "ApplicationId":"256",
      "Position":"101.43.2.193"
    },
    "Name":"umer.nalla"
  }

        
        Copy
    

    	
            
{
"Domain":"Login",
 "Elements":{
   "MaxMsgSize":61430,
   "PingTimeout":30
 },
 "ID":1,
 "Key":{
   "Elements":{
     "AllowSuspectData":1,
     "ApplicationId":"256",
     "SupportViewRequests":1
   },
   "Name":"umer.nalla"
 },
 "State":{
   "Data":"Ok",
   "Stream":"Open",
   "Text":"Login accepted by host centos7-2."
 },
 "Type":"Refresh"
}

        
        Copy
    

    	
            
{
   "ID": 1,
   "Type": "Status",
   "Domain": "Login",
   "Key": {
     "Name": "fred"
   },
   "State": {
     "Stream": "Closed",
     "Data": "Suspect",
     "Code": "UserUnknownToPermSys",
     "Text": "fred, unknown to system."
   }
 }

        
        Copy
    

    	
            
def on_message(ws, message):
    """ Called when message received, parse message into JSON for processing """
    print("RECEIVED: ")
    message_json = json.loads(message)
    print(json.dumps(message_json, sort_keys=True, indent=2, separators=(',', ':')))
    for singleMsg in message_json:
        process_message(ws, singleMsg)

        
        Copy
    

    	
            
def process_message(ws, message_json):
    global shutdown_app
 
    """ Extract Message Type and Domain"""
    message_type = message_json['Type']
    if 'Domain' in message_json:
        message_domain = message_json['Domain']
    # check for a Login Refresh response to confirm successful login
    if message_type == "Refresh" and message_domain == "Login":
        global logged_in
        logged_in = True
        print ("LOGGED IN")
    elif message_type == "Ping":
        pong_json = { 'Type':'Pong' }
        ws.send(json.dumps(pong_json))
        print("SENT:")
        print(json.dumps(pong_json, sort_keys=True, indent=2, separators=(',', ':')))
    elif message_type == "Status" and message_domain == "Login":
        # A Login Status message usually indicates a problem - so report it and shutdown
        if message_json['State']['Stream'] != "Open" or message_json['State']['Data'] != "Ok":
            print("LOGIN REQUEST REJECTED.")
            shutdown_app = True
    elif message_type == "Error":
            shutdown_app = True

        
        Copy
    

    	
            
svcname = "NIPROV"  # Service to Post on - Check with your Market Data team for correct value
ric = "UMER.TST"    # RIC to Post / Contribute data to - Check with your Market Data team for Test RICs
 
bid = 22.1          # A few dummy starting values
ask = 24.5
trdprc = 23.3
 
# Use a python dict to store our FieldName(key) + Value pairs
fields = { 'BID' : bid, 'ASK' : ask, 'TRDPRC_1' : trdprc, 'GEN_TEXT16' : 'some text' }
 
# Send our Refresh message to create the RIC or refresh the fields of an existing RIC
send_mp_offstream_post(svcname,ric, fields, True)

        
        Copy
    

    	
            
def send_mp_offstream_post(svc,riccode, fields, refresh = False):
    global post_id
    """ Send an off-stream post message containing market-price content """
    mp_post_json = {
        'ID': 1,
        'Type':'Post',
        'Key': {
            'Service': svc,
            'Name': riccode
        },
        'Ack':True,
        'PostID':post_id,
        'PostUserInfo': {       # Ask you Market Data team if this is mandated by your organisation
            'Address':position,  # Use IP address as the Post User Address.
            'UserID':os.getpid() # Using process ID as the Post User Id - check with your MD team
        },
        'Message': {
            'ID': 0,
            'Type':'Refresh' if refresh else 'Update',
            'Fields':fields
        }
    }

        
        Copy
    

    	
            
{
  "Ack":true,
  "Domain":"MarketPrice",
  "ID":1,
  "Key":{
    "Name":"UMER.TST",
    "Service":"NIPROV"
  },
  "Message":{
    "Fields":{
      "ASK":24.5,
      "BID":22.1,
      "GEN_TEXT16":"some text",
      "TRDPRC_1":23.3
    },
    "ID":0,
    "Solicited":false,
    "State":{
      "Data":"Ok",
      "Stream":"Open"
    },
    "Type":"Refresh"
  },
  "PostID":1,
  "PostUserInfo":{
    "Address":"10.44.12.152",
    "UserID":7408
  },
  "Type":"Post"
}

        
        Copy
    

    	
            
{
    "AckID":1,
    "ID":1,
    "Key":{
      "Name":"UMER.TST",
      "Service":"NIPROV"
    },
    "Type":"Ack"
}

        
        Copy
    

    	
            
{
    "ID": 1,
    "Type": "Ack",
    "AckID": 1,
    "NakCode": "SymbolUnknown",
    "Text": "F44: Unable to find item on post update.",
    "Key": {
      "Service": "NIPROV",
      "Name": "DAVE.TST"
    }
  }

        
        Copy