Skip to content

Manual - Trends

Download ¤

Download trend using config

config_file = ...
cb = ConfigBundle(file_path=config_file)

today = datetime.date.today()
global_request_params = cb.global_config["request"]
parent_folder = cb.global_config["path"]["parent_folder"]
trends_service = _TrendReq(
    hl=global_request_params["hl"],
    tz=global_request_params["tz"],
    timeout=(10, 14),
    requests_args={"headers": get_random_user_agent()},
    proxies=["https://157.245.27.9:3128"],
)

dl = Download(
    parent_folder=parent_folder,
    snapshot_date=today,
    trends_service=trends_service,
)

wait_seconds_min_max = (30, 120)

for c in cb:
    dl(c)
    wait_seconds = random.randint(*wait_seconds_min_max)
    logger.info(f"Waiting for {wait_seconds} seconds ...")
    time.sleep(wait_seconds)

Parameters:

Name Type Description Default
parent_folder AnyPath

parent folder for the data

 required
snapshot_date datetime.date

snapshot date for the path

 required
trends_service _TrendReq

trend service

 required
Source code in sm_trendy/use_pytrends/get_trends.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
class Download:
    """Download trend using config

    ```python
    config_file = ...
    cb = ConfigBundle(file_path=config_file)

    today = datetime.date.today()
    global_request_params = cb.global_config["request"]
    parent_folder = cb.global_config["path"]["parent_folder"]
    trends_service = _TrendReq(
        hl=global_request_params["hl"],
        tz=global_request_params["tz"],
        timeout=(10, 14),
        requests_args={"headers": get_random_user_agent()},
        proxies=["https://157.245.27.9:3128"],
    )

    dl = Download(
        parent_folder=parent_folder,
        snapshot_date=today,
        trends_service=trends_service,
    )

    wait_seconds_min_max = (30, 120)

    for c in cb:
        dl(c)
        wait_seconds = random.randint(*wait_seconds_min_max)
        logger.info(f"Waiting for {wait_seconds} seconds ...")
        time.sleep(wait_seconds)
    ```

    :params parent_folder: parent folder for the data
    :param snapshot_date: snapshot date for the path
    :param trends_service: trend service
    """

    def __init__(
        self,
        parent_folder: AnyPath,
        snapshot_date: datetime.date,
        trends_service: _TrendReq,
    ):
        self.parent_folder = parent_folder
        self.snapshot_date = snapshot_date
        self.trends_service = trends_service

    def __call__(self, config: Config):
        """
        :param config: config for the keyword
        """

        trend_params = config.trend_params
        path_params = config.path_params
        target_folder = path_params.path(parent_folder=self.parent_folder)
        sdf = StoreDataFrame(
            target_folder=target_folder, snapshot_date=self.snapshot_date
        )

        logger.info(
            f"keyword: {trend_params.keyword}\n" f"target_path: {target_folder}\n" "..."
        )

        st = SingleTrend(
            trends_service=self.trends_service,
            keyword=trend_params.keyword,
            geo=trend_params.geo,
            timeframe=trend_params.timeframe,
            cat=trend_params.cat,
        )

        sdf.save(st, formats=["csv", "parquet"])
        logger.info(f"Saved to {target_folder}")

__call__(config) ¤

Parameters:

Name Type Description Default
config Config

config for the keyword

 required
Source code in sm_trendy/use_pytrends/get_trends.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def __call__(self, config: Config):
    """
    :param config: config for the keyword
    """

    trend_params = config.trend_params
    path_params = config.path_params
    target_folder = path_params.path(parent_folder=self.parent_folder)
    sdf = StoreDataFrame(
        target_folder=target_folder, snapshot_date=self.snapshot_date
    )

    logger.info(
        f"keyword: {trend_params.keyword}\n" f"target_path: {target_folder}\n" "..."
    )

    st = SingleTrend(
        trends_service=self.trends_service,
        keyword=trend_params.keyword,
        geo=trend_params.geo,
        timeframe=trend_params.timeframe,
        cat=trend_params.cat,
    )

    sdf.save(st, formats=["csv", "parquet"])
    logger.info(f"Saved to {target_folder}")

SingleTrend ¤

Get the trend for one keyword, in one country

trends_service can be instatiated using

trends_service = _TrendReq(hl="en-US", tz=120)

Parameters:

Name Type Description Default
trends_service _TrendReq

pytrends TrendReq

 required
geo str

geo code such as "DE"

 required
timeframe str

the time frame of the trend, see pytrends

 'today 5-y'
cat int

category code, see pytrends for a list.

 0
Source code in sm_trendy/use_pytrends/get_trends.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class SingleTrend:
    """Get the trend for one keyword, in one country

    `trends_service` can be instatiated using

    ```python
    trends_service = _TrendReq(hl="en-US", tz=120)
    ```

    :param trends_service: pytrends TrendReq
    :param geo: geo code such as `"DE"`
    :param timeframe: the time frame of the trend, see pytrends
    :param cat: category code, see pytrends for a list.
    """

    def __init__(
        self,
        trends_service: _TrendReq,
        keyword: str,
        geo: str,
        timeframe: str = "today 5-y",
        cat: int = 0,
    ):
        self.trends_service = trends_service
        self.keyword = keyword
        self.cat = cat
        self.geo = geo
        self.timeframe = timeframe

        self.timestamp = datetime.datetime.now()

    @cached_property
    def dataframe(self) -> pd.DataFrame:
        """
        Build the dataframe

        :param keyword: keyword to be searched
        """

        if not isinstance(self.keyword, str):
            raise Exception(f"Requires str as keyword input, got {type(self.keyword)}")

        logger.debug(f"building payload for {self.keyword} ...")
        self.trends_service.build_payload(
            [self.keyword], cat=self.cat, timeframe=self.timeframe, geo=self.geo
        )

        logger.debug(f"Downloading trend for {self.keyword} ...")
        df = self.trends_service.interest_over_time()

        return df

    @cached_property
    def metadata(self) -> Dict[str, Union[str, int]]:
        return {
            "timestamp": self.timestamp.isoformat(),
            "keyword": self.keyword,
            "geo": self.geo,
            "timeframe": self.timeframe,
            "cat": self.cat,
        }

dataframe: pd.DataFrame cached property ¤

Build the dataframe

Parameters:

Name Type Description Default
keyword

keyword to be searched

 required