some intrinio updates

This commit is contained in:
Danglewood
2024-12-11 12:39:39 -08:00
parent f49a3536fa
commit 0834fc458c
4 changed files with 72 additions and 28 deletions

View File

@@ -83,6 +83,10 @@ class IntrinioWebSocketData(WebSocketData):
default=None,
description="The condition attached to the trade or quote.",
)
is_darkpool: Optional[bool] = Field(
default=None,
description="Flag if the trade is reported from an unlit venue.",
)
@field_validator("date", mode="before", check_fields=False)
@classmethod
@@ -115,23 +119,26 @@ class IntrinioWebSocketData(WebSocketData):
def _validate_conditions(cls, values):
"""Validate the exchange."""
new_values = values.copy()
conditions = new_values.pop("condition", None)
trade_type = new_values.get("type")
if trade_type == "trade":
conditions = new_values.pop("condition", None)
if trade_type == "trade":
if not conditions:
return new_values
new_conditions = []
conditions = conditions.replace(" ", "")
for char in range(len(conditions)):
if trade_type == "trade":
new_conditions.append(
TRADE_CONDITIONS.get(conditions[char], conditions[char])
)
new_values["conditions"] = "; ".join(new_conditions)
else:
new_values["conditions"] = conditions if conditions else None
return new_values

View File

@@ -18,24 +18,25 @@ SOURCES = Literal[
]
VENUES = {
"A": "NYSE MKT LLC",
"B": "NASDAQ OMX BX, Inc.",
"C": "National Stock Exchange Inc. (NSX)",
"A": "NYSE MKT",
"B": "NASDAQ OMX BX",
"C": "National Stock Exchange",
"D": "FINRA ADF",
"I": "International Securities Exchange, LLC",
"J": "Bats EDGA Exchange, INC",
"K": "Bats EDGX Exchange, Inc.",
"M": "Chicago Stock Exchange, Inc. (CHX)",
"N": "New York Stock Exchange LLC",
"P": "NYSE Arca, Inc.",
"I": "International Securities Exchange",
"J": "Bats EDGA Exchange",
"K": "Bats EDGX Exchange",
"L": "Long-term Stock Exchange",
"M": "Chicago Stock Exchange",
"N": "New York Stock Exchange",
"P": "NYSE Arca",
"S": "Consolidated Tape System",
"T": "NASDAQ (Tape A, B securities)",
"Q": "NASDAQ (Tape C securities)",
"V": "The Investors' Exchange, LLC (IEX)",
"W": "Chicago Broad Options Exchange, Inc. (CBOE)",
"X": "NASDAQ OMX PSX, Inc. LLC",
"Y": "Bats BYX Exchange, Inc.",
"Z": "Bats BZX Exchange, Inc.",
"V": "The Investors Exchange",
"W": "Cboe",
"X": "NASDAQ OMX PSX",
"Y": "Bats BYX Exchange",
"Z": "Bats BZX Exchange",
"U": "Other OTC Markets",
}

View File

@@ -11,11 +11,13 @@
# This file is a slightly modified version of the original file from the Intrinio Python SDK.
import json
import logging
import queue
import struct
import sys
import threading
import time
from typing import Any, Dict, Optional
@@ -67,7 +69,7 @@ class Quote:
self.timestamp = timestamp
self.subprovider = subprovider
self.market_center = market_center
self.condition = str(condition.strip().replace(" ", ""))
self.condition = condition
def __str__(self):
"""Return string representation of the quote."""
@@ -89,6 +91,30 @@ class Quote:
+ str(self.condition)
)
def is_darkpool(self):
"""Return True if the trade is a dark pool trade."""
return (
not self.market_center
or self.market_center in ("D", "E", "\x00")
or self.market_center.strip() == ""
)
def to_json(self):
return json.dumps(
dict(
symbol=self.symbol,
type=self.type,
price=self.price,
size=self.size,
total_volume=None,
timestamp=self.timestamp,
subprovider=self.subprovider,
market_center=self.market_center,
condition=self.condition,
is_darkpool=self.is_darkpool() if self.is_darkpool() else None,
)
)
class Trade:
"""Intrinio Realtime Stocks Trade."""
@@ -140,6 +166,22 @@ class Trade:
or self.market_center.strip() == ""
)
def to_json(self):
return json.dumps(
dict(
symbol=self.symbol,
type="trade",
price=self.price,
size=self.size,
total_volume=self.total_volume,
timestamp=self.timestamp,
subprovider=self.subprovider,
market_center=self.market_center,
condition=self.condition,
is_darkpool=self.is_darkpool() if self.is_darkpool() else None,
)
)
class IntrinioRealtimeClient:
"""Intrinio Realtime Stocks Client."""
@@ -160,7 +202,7 @@ class IntrinioRealtimeClient:
self.password = options.get("password")
self.provider = options.get("provider")
self.ipaddress = options.get("ipaddress")
self.tradesonly = None
self.tradesonly = options.get("tradesonly")
self.bypass_parsing = options.get("bypass_parsing", False)
if "channels" in options:

View File

@@ -6,10 +6,11 @@ import asyncio
import json
import signal
import sys
from typing import Any
from openbb_core.provider.utils.helpers import run_async
from openbb_intrinio.models.websocket_connection import IntrinioWebSocketData
from openbb_intrinio.utils.stocks_client import IntrinioRealtimeClient, Trade
from openbb_intrinio.utils.stocks_client import IntrinioRealtimeClient
from openbb_websockets.helpers import (
MessageQueue,
get_logger,
@@ -28,16 +29,9 @@ CONNECT_KWARGS = kwargs.pop("connect_kwargs", {})
async def process_message(message):
"""Process the message and write to the database."""
result: dict = {}
message = json.loads(message) if isinstance(message, str) else message
is_trade = isinstance(message, Trade)
if hasattr(message, "__dict__"):
message = message.__dict__
if is_trade:
message["type"] = "trade"
result: Any = None
try:
result = IntrinioWebSocketData.model_validate(message)
result = IntrinioWebSocketData.model_validate_json(message.to_json())
result = ( # type: ignore
{}
if result.exchange == "!" or result.price == 0 # type: ignore