"""
Topic Analysis Algorithms
==========================
This module provides functions for analyzing topic-related dynamics in YSocial simulations.
These functions help understand how topics spread, how quickly they are adopted,
and when engagement peaks occur.
All public helpers in this module are implemented on top of the topic lifecycle
summaries exposed by :class:`ysights.models.YDataHandler`.
Example:
Basic usage of topic analysis functions::
from ysights import YDataHandler
from ysights.algorithms import topics
# Initialize data handler
ydh = YDataHandler('path/to/database.db')
# Analyze topic spread
spread = topics.topic_spread(ydh)
# Calculate adoption rates
rates = topics.adoption_rate(ydh)
print(f"Adoption rates available for {len(rates)} topics")
"""
from ysights import YDataHandler
def _topic_ids(YDH: YDataHandler):
if YDH is None:
raise ValueError("A YDataHandler instance is required.")
if not YDH.supports_feature("topics"):
raise ValueError("Topic analysis is not available in this dataset.")
rows = YDH.custom_query(
"SELECT DISTINCT topic_id FROM post_topics ORDER BY topic_id"
)
return [row[0] for row in rows]
[docs]
def topic_spread(YDH: YDataHandler):
"""
Analyze the spread of topics across the social network.
This function will analyze how topics diffuse through the network over time,
identifying patterns of information spread and influence.
:param YDH: YDataHandler instance for database operations
:type YDH: YDataHandler
:return: Topic lifecycle summaries keyed by topic identifier.
:rtype: dict
Example::
from ysights import YDataHandler
from ysights.algorithms.topics import topic_spread
ydh = YDataHandler('path/to/database.db')
# Analyze topic spread
results = topic_spread(ydh)
print(results)
See Also:
:func:`adoption_rate`: Calculate topic adoption rates
:func:`peak_engagement_time`: Find peak engagement times
"""
topic_ids = _topic_ids(YDH)
return {topic_id: YDH.topic_lifecycle(topic_id) for topic_id in topic_ids}
[docs]
def adoption_rate(YDH: YDataHandler):
"""
Calculate the adoption rate of topics over time.
This function will measure how quickly agents adopt and engage with
different topics in the simulation, providing insights into topic
popularity and virality.
:param YDH: YDataHandler instance for database operations
:type YDH: YDataHandler
:return: Topic adoption rates keyed by topic identifier.
:rtype: dict
Example::
from ysights import YDataHandler
from ysights.algorithms.topics import adoption_rate
ydh = YDataHandler('path/to/database.db')
# Calculate adoption rates
rates = adoption_rate(ydh)
print(f"Average adoption rate: {sum(rates.values()) / len(rates):.3f}")
See Also:
:func:`topic_spread`: Analyze topic spread patterns
:func:`peak_engagement_time`: Find peak engagement times
"""
topic_ids = _topic_ids(YDH)
return {
topic_id: YDH.topic_lifecycle(topic_id)["adoption_rate"]
for topic_id in topic_ids
}
[docs]
def peak_engagement_time(YDH: YDataHandler):
"""
Identify peak engagement times for topics.
This function will determine when topics receive the most attention
and engagement from agents, helping to understand temporal patterns
in topic dynamics.
:param YDH: YDataHandler instance for database operations
:type YDH: YDataHandler
:return: Peak engagement periods keyed by topic identifier.
:rtype: dict
Example::
from ysights import YDataHandler
from ysights.algorithms.topics import peak_engagement_time
ydh = YDataHandler('path/to/database.db')
# Find peak engagement times
peaks = peak_engagement_time(ydh)
for topic, peak_time in peaks.items():
print(f"Topic {topic} peaked at {peak_time}")
See Also:
:func:`topic_spread`: Analyze topic spread patterns
:func:`adoption_rate`: Calculate topic adoption rates
"""
topic_ids = _topic_ids(YDH)
return {
topic_id: YDH.topic_lifecycle(topic_id)["peak_period"] for topic_id in topic_ids
}