量化A股舆情：基于Kafka+Faust的实时新闻流解析

pip install python-kafka

# encoding utf-8
from kafka import KafkaConsumer # 注意这里是kafka，不是python-kafka
import json

def get_news_stream(start_from_newest=True):
    """
    :param start_from_newest: False, 从上次结束消费的地方接收信息，True, 从最新的消息开始接收
    :return:
    """
    consumer = KafkaConsumer('TOPIC_NAME',
                             group_id='GROUP_ID',
                             enable_auto_commit=True,
                             auto_commit_interval_ms=1000,
                             auto_offset_reset='latest',
                             security_protocol='SASL_PLAINTEXT',
                             sasl_mechanism="SCRAM-SHA-512",
                             sasl_plain_username="USERNAME",
                             sasl_plain_password="PASSWORD",
                             bootstrap_servers=['broker1.xxxxxx.com:19092',
 'broker2.xxxxxx.com:19092']
                             )
    if start_from_newest:
        consumer.poll(10)
        consumer.seek_to_end()
    for msg in consumer:
        print(json.loads(msg.value)) # 可以加入对于新闻处理的逻辑，这里只是简单Print

if __name__ == '__main__':
    get_news_stream()

pip install faust-streaming # 注意不是pip install faust

# news_stream.py
import faust

# 定义App
app = faust.App('GROUP_ID',
                value_serializer='json',
                topic_allow_declare=False,
                topic_disable_leader=True,
                broker=['kafka://broker1.xxxxxx.com:19092',
                        'kafka://broker2.xxxxxx.com:19092'],
                broker_credentials=faust.SASLCredentials(
                    username='USERNAME',
                    password='PASSWORD',
                    mechanism='SCRAM-SHA-512')
               )
               

# 订阅Topic
news_kafka_topic = app.topic('TOPIC_NAME')

# 针对Topic的处理函数
@app.agent(news_kafka_topic)
async def process(stream) -> None:
    async for event in news_items.events():
        print('News:{}'.format(event.message.timestamp))
        print(event.value)

if __name__ == '__main__':
    app.main()

faust -A news_stream woker -l info #news_stream

# encoding=utf-8
from abc import ABC
import faust
import smtplib
from email.mime.text import MIMEText

class NewsItem(faust.Record, ABC, serializer='json'):
    newsInfo: dict
    newsTags: list
    emotionInfos: list

def send_mail(content):
    # 邮件构建
    subject = "负面新闻预警"  # 邮件标题
    sender = "*********@163.com"  # 发送方
    recver = "*******@163.com"  # 接收方
    password = "*****"
    message = MIMEText(content, "plain", "utf-8")
    # content 发送内容 "plain"文本格式 utf-8 编码格式
    message['Subject'] = subject # 邮件标题
    message['To'] = recver # 收件人
    message['From'] = sender # 发件人
    smtp = smtplib.SMTP_SSL("smtp.163.com", 994) # 实例化smtp服务器
    smtp.login(sender, password) # 发件人登录
    smtp.sendmail(sender, [recver], message.as_string()) # as_string 对 message 的消息进行了封装
    smtp.close()

def news_alert(news_time, news_value):
    sec_codes = ['300750_SZ_EQ', '002594_SZ_EQ', '600050_SH_EQ', '002230_SZ_EQ']
    news_info = news_value.newsInfo
    tag_info = news_value.newsTags
    emotion_info = news_value.emotionInfos
    for tag in tag_info:
        if tag['entityCode'] in sec_codes and tag['itemRelevance']>=0.5:
            for emotion in emotion_info:
                if emotion['entityCode'] == tag['entityCode'] and emotion['emotionIndicator']==2 and emotion['emotionWeight']>=0.8:
                    content = "{}【负面预警】:{}-{} \n 【新闻摘要】{}".format(news_time, tag['entityCode'], emotion['emotionWeight'], news_info['newsSummary'])
    send_mail(content)

app = faust.App('GROUP_NAME',
                value_serializer='json',
                topic_allow_declare=False,
                topic_disable_leader=True,
                broker=['kafka://xxxx.com:19092',
                        'kafka://yyyy.com:19092'],
                broker_credentials=faust.SASLCredentials(
                    username='USERNAME',
                    password='PASSWORD',
                    mechanism='SCRAM-SHA-512')
                )
news_kafka_topic = app.topic('TOPIC_NAME', value_type=NewsItem)

@app.agent(news_kafka_topic)
async def process(news) -> None:
    async for news_item in news.events():
        news_alert(news_item.message.timestamp, news_item.value)

if __name__ == '__main__':
    app.main()