Implementation:Online ml River Stream TwitchChatStream
| Knowledge Sources | |
|---|---|
| Domains | Online_Learning, Data_Streaming, Social_Media, Twitch |
| Last Updated | 2026-02-08 16:00 GMT |
Overview
Real-time Twitch chat message stream client using IRC protocol for online learning on live gaming community data.
Description
The TwitchChatStream class connects to Twitch's IRC servers to stream live chat messages from specified channels. It handles authentication via OAuth tokens, manages multiple channels simultaneously, responds to server PINGs, and yields structured chat message objects with timestamps, channel names, usernames, and message content. Configurable buffer size and timeout parameters allow tuning for different network conditions.
Usage
Use this for sentiment analysis of gaming communities, toxicity detection, trend analysis, or building adaptive chatbots. Ideal for real-time moderation systems, community behavior analysis, or any application requiring online learning on streaming chat data.
Code Reference
Source Location
- Repository: Online_ml_River
- File: river/stream/twitch_chat_stream.py
Signature
class TwitchChatStream:
def __init__(
self,
nickname: str,
token: str,
channels: list[str],
buffer_size: int = BUFFER_SIZE,
timeout: int = TIMEOUT,
):
...
def __iter__(self) -> Iterator[dict]:
...
Import
from river import stream
I/O Contract
| Parameter | Type | Description |
|---|---|---|
| nickname | str | Your Twitch account nickname |
| token | str | OAuth token from https://twitchapps.com/tmi/ |
| channels | list[str] | List of channel names to monitor (without #) |
| buffer_size | int | IRC buffer size in bytes (default: 2048) |
| timeout | int | Server response timeout in seconds (default: 60) |
Returns (via iteration):
| Field | Type | Description |
|---|---|---|
| dt | datetime.datetime | Message receipt timestamp |
| channel | str | Channel where message was posted |
| username | str | User who posted the message |
| msg | str | The message content |
Usage Examples
from river import stream
# Initialize with credentials and channels
twitch_chat = stream.TwitchChatStream(
nickname="your_twitch_username",
token="oauth:your_token_here",
channels=["shroud", "ninja", "pokimane"]
)
# Stream chat messages
for item in twitch_chat:
print(f"[{item['dt']}] {item['channel']}/{item['username']}: {item['msg']}")
# Example with sentiment analysis
from river import feature_extraction, naive_bayes, compose
# Build a simple sentiment classifier
model = compose.Pipeline(
('tokenize', feature_extraction.BagOfWords()),
('classify', naive_bayes.BernoulliNB())
)
# Labels dict for manual labeling (in practice, use pre-labeled data)
sentiment_labels = {
'positive': ['!', 'love', 'great', 'awesome', 'pog'],
'negative': ['bad', 'hate', 'worst', 'terrible']
}
def simple_sentiment(msg):
msg_lower = msg.lower()
for word in sentiment_labels['positive']:
if word in msg_lower:
return 'positive'
for word in sentiment_labels['negative']:
if word in msg_lower:
return 'negative'
return None
# Stream and learn
for item in twitch_chat:
msg = item['msg']
label = simple_sentiment(msg)
if label:
# Predict before learning
pred = model.predict_one(msg)
# Learn from labeled message
model.learn_one(msg, label)
if pred != label:
print(f"Misclassified: {msg[:50]}... (predicted: {pred}, actual: {label})")
# Message structure example:
# {
# 'dt': datetime.datetime(2022, 9, 14, 10, 33, 37, 989560),
# 'channel': 'shroud',
# 'username': 'viewer123',
# 'msg': 'that play was insane!'
# }