Guides

Build a daily horoscope Telegram bot

This bot replies to /horoscope <sign> with today's prediction for that zodiac sign. The bot can also push a subscriber's horoscope automatically every morning.

The bot runs in Python against AstrologyAPI's sun sign endpoint. See the full list of guides for other ways to build with the API.

What you’ll build

A /horoscope <sign> command that replies with today’s prediction right away.
A /subscribe <sign> and /unsubscribe pair that registers a chat for a daily push.
A scheduled job that sends each subscriber their sign’s horoscope every morning.

Prerequisites

Python 3.10 or later.
A Telegram bot token. Message @BotFather on Telegram, send /newbot, and save the token it returns.
An AstrologyAPI user ID and API key from your dashboard.

The horoscope endpoint

The bot calls one endpoint: POST /v1/sun_sign_prediction/daily/{zodiacName}. The sign is a path parameter, so the sign goes on the URL path, not in the body.

Auth is HTTP Basic: your user ID is the username, your API key is the password.

Field	Type	Required	Description
`zodiacName`	string (path)	Yes	One of: aries, taurus, gemini, cancer, leo, virgo, libra, scorpio, sagittarius, capricorn, aquarius, pisces.
`timezone`	float (body)	No	UTC offset in hours. Defaults to 5.5 if omitted.

Every response carries the same six fields, each a prediction paragraph for one life area.

Field	Type	Description
`personal_life`	string	Relationships and communication.
`profession`	string	Work and career.
`health`	string	Health and energy.
`emotions`	string	Mood and feelings.
`travel`	string	Trips and movement.
`luck`	string	Chance and timing.

A request and its response look like this. The values shown are illustrative placeholders; the live endpoint returns fresh prediction text each day.

POST https://json.astrologyapi.com/v1/sun_sign_prediction/daily/leo

{
  "timezone": 5.5
}

Response:
{
  "personal_life": "Today's outlook on relationships and communication.",
  "profession": "Today's outlook on work and career.",
  "health": "Today's outlook on health and energy.",
  "emotions": "Today's outlook on mood and feelings.",
  "travel": "Today's outlook on trips and movement.",
  "luck": "Today's outlook on chance and timing."
}

Install the dependencies

pip install "python-telegram-bot[job-queue]" requests

The [job-queue] extra is required because JobQueue depends on APScheduler since python-telegram-bot v20. Without the extra, application.job_queue is None and the daily push cannot run.

Reply to /horoscope

The command handler is an async function. It reads the sign from context.args, validates the sign, fetches the prediction, and replies.

Two helpers do the work. format_prediction turns the six fields into a readable message. get_daily_prediction calls the endpoint. This is a partial snippet; the full script comes later.

# The command handler is an async function.
# It reads the sign, fetches the prediction, and replies.

def format_prediction(data):
    return "\n\n".join(
        [
            f"Personal life: {data['personal_life']}",
            f"Profession: {data['profession']}",
            f"Health: {data['health']}",
            f"Emotions: {data['emotions']}",
            f"Travel: {data['travel']}",
            f"Luck: {data['luck']}",
        ]
    )


def get_daily_prediction(sign):
    response = requests.post(
        f"{API_BASE_URL}/sun_sign_prediction/daily/{sign}",
        auth=(ASTROLOGYAPI_USER_ID, ASTROLOGYAPI_API_KEY),
        json={"timezone": DEFAULT_TIMEZONE},
        timeout=10,
    )
    response.raise_for_status()
    return format_prediction(response.json())


async def horoscope(update, context):
    if not context.args:
        await update.message.reply_text(
            "Usage: /horoscope <sign>. Signs: " + ", ".join(ZODIAC_SIGNS)
        )
        return
    sign = context.args[0].lower()
    if sign not in ZODIAC_SIGNS:
        await update.message.reply_text(
            "Unknown sign. Signs: " + ", ".join(ZODIAC_SIGNS)
        )
        return
    await update.message.reply_text(get_daily_prediction(sign))

Cache the prediction, don’t refetch it

The response for a sign does not change again until the next calendar day. A second call for the same sign on the same day returns the same text.

Cache by sign and date in memory. Skip the network call on a cache hit.

This also matters when several subscribers share a sign. The daily job should hit the API once per sign, not once per subscriber. Ten Leo subscribers cost one request, not ten.

Send it automatically every morning

/subscribe <sign> and /unsubscribe persist a chat_id to sign mapping in a small JSON file. The subscriber list then survives a restart.

application.job_queue.run_daily(...) runs a job once a day. The job reads the file and messages each subscriber.

def load_subscribers():
    if not os.path.exists(SUBSCRIBERS_FILE):
        return {}
    with open(SUBSCRIBERS_FILE) as file:
        return json.load(file)


def save_subscribers(subscribers):
    with open(SUBSCRIBERS_FILE, "w") as file:
        json.dump(subscribers, file)


async def subscribe(update, context):
    if not context.args or context.args[0].lower() not in ZODIAC_SIGNS:
        await update.message.reply_text(
            "Usage: /subscribe <sign>. Signs: " + ", ".join(ZODIAC_SIGNS)
        )
        return
    sign = context.args[0].lower()
    subscribers = load_subscribers()
    subscribers[str(update.effective_chat.id)] = sign
    save_subscribers(subscribers)
    await update.message.reply_text(
        f"Subscribed to {sign}. Sent daily at 08:00 Asia/Kolkata."
    )


async def unsubscribe(update, context):
    subscribers = load_subscribers()
    subscribers.pop(str(update.effective_chat.id), None)
    save_subscribers(subscribers)
    await update.message.reply_text("Unsubscribed.")


async def send_daily_horoscopes(context):
    subscribers = load_subscribers()
    for chat_id, sign in subscribers.items():
        text = get_daily_prediction(sign)
        await context.bot.send_message(chat_id=chat_id, text=text)


# Register the job once, when the app starts.
application.job_queue.run_daily(send_daily_horoscopes, time=DAILY_SEND_TIME)

Complete bot

Here is the full script, assembled from the pieces above. The script handles four jobs.

Reads the bot token, user ID, and API key from environment variables.
Caches one prediction per sign per day.
Persists subscribers to disk.
Registers the daily job.

Set the three environment variables, then run the script.

import json
import os
from datetime import date, time
from zoneinfo import ZoneInfo

import requests
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes

# Read secrets from the environment. Fail fast if any are unset.
TELEGRAM_BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
ASTROLOGYAPI_USER_ID = os.environ["ASTROLOGYAPI_USER_ID"]
ASTROLOGYAPI_API_KEY = os.environ["ASTROLOGYAPI_API_KEY"]

API_BASE_URL = "https://json.astrologyapi.com/v1"
DEFAULT_TIMEZONE = 5.5
SUBSCRIBERS_FILE = "subscribers.json"
DAILY_SEND_TIME = time(hour=8, minute=0, tzinfo=ZoneInfo("Asia/Kolkata"))

ZODIAC_SIGNS = [
    "aries",
    "taurus",
    "gemini",
    "cancer",
    "leo",
    "virgo",
    "libra",
    "scorpio",
    "sagittarius",
    "capricorn",
    "aquarius",
    "pisces",
]

# Cache one prediction per sign per day, keyed by sign.
_prediction_cache = {}


def load_subscribers():
    if not os.path.exists(SUBSCRIBERS_FILE):
        return {}
    with open(SUBSCRIBERS_FILE) as file:
        return json.load(file)


def save_subscribers(subscribers):
    with open(SUBSCRIBERS_FILE, "w") as file:
        json.dump(subscribers, file)


def format_prediction(data):
    return "\n\n".join(
        [
            f"Personal life: {data['personal_life']}",
            f"Profession: {data['profession']}",
            f"Health: {data['health']}",
            f"Emotions: {data['emotions']}",
            f"Travel: {data['travel']}",
            f"Luck: {data['luck']}",
        ]
    )


def get_daily_prediction(sign):
    today = date.today().isoformat()
    cached = _prediction_cache.get(sign)
    if cached and cached["date"] == today:
        return cached["text"]

    # This is a synchronous, blocking call made from async handlers.
    # High-traffic bots should wrap it in asyncio.to_thread(...) instead.
    response = requests.post(
        f"{API_BASE_URL}/sun_sign_prediction/daily/{sign}",
        auth=(ASTROLOGYAPI_USER_ID, ASTROLOGYAPI_API_KEY),
        json={"timezone": DEFAULT_TIMEZONE},
        timeout=10,
    )
    response.raise_for_status()
    text = format_prediction(response.json())
    _prediction_cache[sign] = {"date": today, "text": text}
    return text


async def start(update, context):
    await update.message.reply_text(
        "Commands:\n"
        "/horoscope <sign> — today's prediction\n"
        "/subscribe <sign> — daily push at 08:00 Asia/Kolkata\n"
        "/unsubscribe — stop the daily push"
    )


async def horoscope(update, context):
    if not context.args:
        await update.message.reply_text(
            "Usage: /horoscope <sign>. Signs: " + ", ".join(ZODIAC_SIGNS)
        )
        return
    sign = context.args[0].lower()
    if sign not in ZODIAC_SIGNS:
        await update.message.reply_text(
            "Unknown sign. Signs: " + ", ".join(ZODIAC_SIGNS)
        )
        return
    await update.message.reply_text(get_daily_prediction(sign))


async def subscribe(update, context):
    if not context.args or context.args[0].lower() not in ZODIAC_SIGNS:
        await update.message.reply_text(
            "Usage: /subscribe <sign>. Signs: " + ", ".join(ZODIAC_SIGNS)
        )
        return
    sign = context.args[0].lower()
    subscribers = load_subscribers()
    subscribers[str(update.effective_chat.id)] = sign
    save_subscribers(subscribers)
    await update.message.reply_text(
        f"Subscribed to {sign}. Sent daily at 08:00 Asia/Kolkata."
    )


async def unsubscribe(update, context):
    subscribers = load_subscribers()
    subscribers.pop(str(update.effective_chat.id), None)
    save_subscribers(subscribers)
    await update.message.reply_text("Unsubscribed.")


async def send_daily_horoscopes(context):
    subscribers = load_subscribers()
    for chat_id, sign in subscribers.items():
        text = get_daily_prediction(sign)
        await context.bot.send_message(chat_id=chat_id, text=text)


def main():
    application = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("horoscope", horoscope))
    application.add_handler(CommandHandler("subscribe", subscribe))
    application.add_handler(CommandHandler("unsubscribe", unsubscribe))
    application.job_queue.run_daily(send_daily_horoscopes, time=DAILY_SEND_TIME)
    application.run_polling()


if __name__ == "__main__":
    main()

Three related endpoints follow the same shape if you need them.

sun_sign_prediction/daily/next/:zodiacName returns the next day’s prediction.
sun_sign_prediction/daily/previous/:zodiacName returns the previous day’s prediction.
sun_sign_consolidated/daily/:zodiacName returns a consolidated prediction.

Checklist

Never commit the bot token or API key. Load them from environment variables.
Persist subscribers to disk so a restart doesn’t lose the list.
Cache per sign per day rather than calling the endpoint on every message or every subscriber.

To send a branded PDF instead of a plain text reply, see Generate your first PDF report. Before you put this bot in front of real users, work through the production checklist.