Streamz with websocket not steaming any data
Chưa có ai nhận issue này.
Đánh giá
- Độ khó
- 3/5
- Thời gian dự kiến
- 1-2 ngày
- Mức phù hợp với người mới
- 25/100
- Loại issue
- Lỗi
- Độ rõ ràng
- Cần làm rõ
- Mức độ hoạt động
- Đình trệ
- Công nghệ
- pandas, python
- Lĩnh vực
- stream-processing
Hướng nghiên cứu
Start at the WebSocket callbacks, update_dataframe, source.emit, and source.map(print_result) shown in the issue. Trace whether messages reach source.emit and whether the mapped consumer runs; done means identifying the missing connection or usage detail that prevents received data from being printed.
Do mô hình lập chỉ mục viết ra từ nội dung của issue.
Mô tả
I am trying to figure out a right way of streaming data using streamz with websocket. My streaming data is loaded using websocket but not streaming anything.
import config
import websocket, json
import pandas as pd
from streamz.dataframe import DataFrame
from streamz import Stream
import time
ticker_list = ["SCHG"]
# Create DataFrame with specified columns and ticker_list as index
df = pd.DataFrame(columns=['Bid', 'Ask', 'Time'], index=ticker_list)
source = Stream()
def update_dataframe(json_message, df):
ticker_data = json_message[0]
ticker_symbol = ticker_data['S']
df.loc[ticker_symbol, 'Bid'] = ticker_data['bp']
df.loc[ticker_symbol, 'Ask'] = ticker_data['ap']
df.loc[ticker_symbol, 'Time'] = pd.to_datetime(ticker_data['t'], unit='s').tz_localize('UTC').tz_convert('America/New_York').strftime('%Y-%m-%d %H:%M:%S%z')
source.emit(df.copy()) # Emit the updated DataFrame to the stream
def on_open(ws):
print("opened connection")
# Authentication
auth_data = {"action":"auth","key": config.API_KEY,"secret": config.SECRET_KEY}
# Subscribe
ws.send(json.dumps(auth_data))
listen_message = {"action":"subscribe","quotes":ticker_list}
ws.send(json.dumps(listen_message))
print('subscribed')
def on_message(ws, message):
json_message = json.loads(message)
# Update the DataFrame and emit it to the stream
update_dataframe(json_message, df)
def on_close(ws):
print("closed connection")
def print_result(x):
print(x)
sink = source.map(print_result) # Set up a sink to consume and process the emitted data
socket = "wss://stream.data.alpaca.markets/v2/iex"
ws = websocket.WebSocketApp(socket, on_open=on_open, on_message=on_message, on_close=on_close)
ws.run_forever()
- Ngôn ngữ chính
- Python
- Star
- 1.3k
- Fork
- 149
- Merge trung bình
- 17 giờ 39 phút
- Pull request đã merge (30 ngày)
- 1
Hướng dẫn đóng góp
Bắt đầu từ đâu
- Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
- Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
- Fork repository và làm thay đổi trên một nhánh.
- Mở pull request có tham chiếu số hiệu của issue.
Issue khác của python-streamz/streamz
-
pkg_resources warning Đang mở
Độ khó 3/5 1-2 ngày Mức phù hợp với người mới 35/100
python-streamz/streamz#481 · 4 bình luận ·
-
Combining the streamz.Stream.filenames() and streamz.Stream.from_textfile() using dask scatter? Đang mở
Độ khó 5/5 Hơn một tuần Mức phù hợp với người mới 25/100
python-streamz/streamz#480 · 2 bình luận ·
-
Compile the code into c++ Đang mở
Độ khó 5/5 Hơn một tuần Mức phù hợp với người mới 20/100
python-streamz/streamz#479 · 1 bình luận ·
-
Độ khó 5/5 Hơn một tuần Mức phù hợp với người mới 20/100
python-streamz/streamz#478 · 6 bình luận ·
-
Độ khó 4/5 3-5 ngày Mức phù hợp với người mới 10/100
python-streamz/streamz#476 · 17 bình luận · 2 reaction ·
Tất cả issue của python-streamz/streamz
Issue tương tự
-
[Bug] reef-hermes tells me to resume with hermes --resume, which does not work from my shell Đang mởarea: harness bug status: needs-triage
Độ khó 2/5 1-3 giờ Mức phù hợp với người mới 75/100
Human-Agent-Society/reef#625 ·
-
Độ khó 2/5 1-3 giờ Mức phù hợp với người mới 70/100
-
Độ khó 1/5 Dưới một giờ Mức phù hợp với người mới 80/100
learningequality/kolibri#15351 · 2 bình luận ·
-
Độ khó 2/5 1-3 giờ Mức phù hợp với người mới 75/100
-
Name consistency Đang mở
Độ khó 2/5 1-3 giờ Mức phù hợp với người mới 75/100
eellak/triplestore#65 · 1 bình luận ·