candle close alert
This commit is contained in:
parent
30a7f1b68c
commit
dd03cae0f3
10 changed files with 328 additions and 60 deletions
|
|
@ -9,6 +9,7 @@ import (
|
|||
"sync"
|
||||
|
||||
"gitea.computernetthings.ru/yash/crypto_alert_bot/internal/entities"
|
||||
"gitea.computernetthings.ru/yash/crypto_alert_bot/internal/provider"
|
||||
tgbotapi "github.com/go-telegram-bot-api/telegram-bot-api/v5"
|
||||
"github.com/shopspring/decimal"
|
||||
)
|
||||
|
|
@ -50,6 +51,7 @@ type flowStep string
|
|||
|
||||
const (
|
||||
stepAddAlertPrice flowStep = "add_alert_price"
|
||||
stepAddAlertAwaitType flowStep = "add_alert_await_type" // price entered, waiting for type callback
|
||||
stepEditAlertPrice flowStep = "edit_alert_price"
|
||||
)
|
||||
|
||||
|
|
@ -57,9 +59,27 @@ type userState struct {
|
|||
step flowStep
|
||||
instrument entities.Instrument // set during add_alert flow
|
||||
currentPrice *entities.Price // fetched when instrument is selected
|
||||
targetPrice decimal.Decimal // set after price is entered, before type selection
|
||||
alertID entities.AlertID // set during edit flow
|
||||
}
|
||||
|
||||
// alertTimeframe pairs a KlineInterval with its display label.
|
||||
type alertTimeframe struct {
|
||||
interval provider.KlineInterval
|
||||
label string
|
||||
}
|
||||
|
||||
// offeredTimeframes are the timeframes shown to the user when creating a candle-close alert.
|
||||
var offeredTimeframes = []alertTimeframe{
|
||||
{provider.Kline1m, "1m"},
|
||||
{provider.Kline5m, "5m"},
|
||||
{provider.Kline15m, "15m"},
|
||||
{provider.Kline1H, "1H"},
|
||||
{provider.Kline4H, "4H"},
|
||||
{provider.Kline1D, "1D"},
|
||||
{provider.Kline1W, "1W"},
|
||||
}
|
||||
|
||||
// Bot is the Telegram bot handling all user interactions.
|
||||
type Bot struct {
|
||||
api *tgbotapi.BotAPI
|
||||
|
|
@ -122,11 +142,12 @@ func (b *Bot) NotifyAlert(ctx context.Context, userID entities.UserID, alert *en
|
|||
}
|
||||
|
||||
text := fmt.Sprintf(
|
||||
"Alert triggered!\n\n%s/%s %s %s\nCurrent price: %s",
|
||||
"Alert triggered!\n\n%s/%s %s %s%s\nClose price: %s",
|
||||
alert.Instrument.BaseCurrency,
|
||||
alert.Instrument.QuoteCurrency,
|
||||
alert.Condition,
|
||||
formatCondition(alert.Condition),
|
||||
alert.Price.String(),
|
||||
formatTimeframeSuffix(alert),
|
||||
currentPrice.String(),
|
||||
)
|
||||
msg := tgbotapi.NewMessage(int64(user.TelegramID), text)
|
||||
|
|
@ -137,6 +158,30 @@ func (b *Bot) NotifyAlert(ctx context.Context, userID entities.UserID, alert *en
|
|||
return err
|
||||
}
|
||||
|
||||
// formatCondition returns a human-readable alert condition label.
|
||||
func formatCondition(c entities.AlertCondition) string {
|
||||
switch c {
|
||||
case entities.AlertConditionAbove:
|
||||
return "above"
|
||||
case entities.AlertConditionBelow:
|
||||
return "below"
|
||||
case entities.AlertConditionCloseAbove:
|
||||
return "close above"
|
||||
case entities.AlertConditionCloseBelow:
|
||||
return "close below"
|
||||
default:
|
||||
return string(c)
|
||||
}
|
||||
}
|
||||
|
||||
// formatTimeframeSuffix returns " (4H)" for candle-close alerts, empty string otherwise.
|
||||
func formatTimeframeSuffix(alert *entities.Alert) string {
|
||||
if alert.Timeframe == "" {
|
||||
return ""
|
||||
}
|
||||
return fmt.Sprintf(" (%s)", alert.Timeframe)
|
||||
}
|
||||
|
||||
// --- Routing ---
|
||||
|
||||
func (b *Bot) handleUpdate(ctx context.Context, update tgbotapi.Update) {
|
||||
|
|
@ -175,6 +220,8 @@ func (b *Bot) handleMessage(ctx context.Context, msg *tgbotapi.Message) {
|
|||
switch state.step {
|
||||
case stepAddAlertPrice:
|
||||
b.handleAddAlertPrice(ctx, tgID, chatID, msg.Text, state)
|
||||
case stepAddAlertAwaitType:
|
||||
b.send(chatID, "Please select the alert type using the buttons above.")
|
||||
case stepEditAlertPrice:
|
||||
b.handleEditAlertPrice(ctx, tgID, chatID, msg.Text, state)
|
||||
default:
|
||||
|
|
@ -253,6 +300,16 @@ func (b *Bot) handleCallback(ctx context.Context, cb *tgbotapi.CallbackQuery) {
|
|||
alertID := entities.AlertID(rest[:idx])
|
||||
page, _ := strconv.Atoi(rest[idx+1:])
|
||||
b.handleAlertSelect(ctx, chatID, messageID, alertID, page)
|
||||
|
||||
case data == "alert_type:crossing":
|
||||
b.handleAlertTypeCrossing(ctx, tgID, chatID)
|
||||
|
||||
case data == "alert_type:close":
|
||||
b.handleAlertTypeClose(tgID, chatID)
|
||||
|
||||
case strings.HasPrefix(data, "alert_timeframe:"):
|
||||
tf := provider.KlineInterval(strings.TrimPrefix(data, "alert_timeframe:"))
|
||||
b.handleAlertTimeframe(ctx, tgID, chatID, tf)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -362,12 +419,13 @@ func buildAlertsPage(alerts []entities.Alert, page int) (string, tgbotapi.Inline
|
|||
var sb strings.Builder
|
||||
fmt.Fprintf(&sb, "Your active alerts (page %d/%d):\n\n", page+1, totalPages)
|
||||
for i, alert := range pageAlerts {
|
||||
fmt.Fprintf(&sb, "%d. %s/%s %s %s\n",
|
||||
fmt.Fprintf(&sb, "%d. %s/%s %s %s%s\n",
|
||||
start+i+1,
|
||||
alert.Instrument.BaseCurrency,
|
||||
alert.Instrument.QuoteCurrency,
|
||||
alert.Condition,
|
||||
formatCondition(alert.Condition),
|
||||
alert.Price.String(),
|
||||
formatTimeframeSuffix(&alert),
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -438,11 +496,12 @@ func (b *Bot) handleAlertSelect(ctx context.Context, chatID int64, messageID int
|
|||
return
|
||||
}
|
||||
|
||||
text := fmt.Sprintf("%s/%s\nCondition: %s %s",
|
||||
text := fmt.Sprintf("%s/%s\nCondition: %s %s%s",
|
||||
alert.Instrument.BaseCurrency,
|
||||
alert.Instrument.QuoteCurrency,
|
||||
alert.Condition,
|
||||
formatCondition(alert.Condition),
|
||||
alert.Price.String(),
|
||||
formatTimeframeSuffix(alert),
|
||||
)
|
||||
|
||||
kb := tgbotapi.NewInlineKeyboardMarkup(tgbotapi.NewInlineKeyboardRow(
|
||||
|
|
@ -507,8 +566,7 @@ func (b *Bot) handleInstrumentSelected(ctx context.Context, tgID entities.Telegr
|
|||
))
|
||||
}
|
||||
|
||||
// handleAddAlertPrice parses the target price, auto-determines the condition from the current
|
||||
// market price (target >= ask → above, target < ask → below), and creates the alert.
|
||||
// handleAddAlertPrice parses the target price, stores it in state, then asks for the alert type.
|
||||
func (b *Bot) handleAddAlertPrice(ctx context.Context, tgID entities.TelegramID, chatID int64, text string, state *userState) {
|
||||
target, err := decimal.NewFromString(strings.TrimSpace(text))
|
||||
if err != nil || !target.IsPositive() {
|
||||
|
|
@ -516,27 +574,40 @@ func (b *Bot) handleAddAlertPrice(ctx context.Context, tgID entities.TelegramID,
|
|||
return
|
||||
}
|
||||
|
||||
// Use the price already fetched at instrument-selection time; re-fetch only if missing.
|
||||
currentPrice := state.currentPrice
|
||||
if currentPrice == nil {
|
||||
currentPrice, err = b.provider.Price(ctx, state.instrument)
|
||||
// Re-fetch price if missing (e.g. failed earlier).
|
||||
if state.currentPrice == nil {
|
||||
state.currentPrice, err = b.provider.Price(ctx, state.instrument)
|
||||
if err != nil {
|
||||
b.log.Error("failed to fetch current price", "instrument", state.instrument.ID, "err", err)
|
||||
b.send(chatID, "Could not fetch current price. Please try again:")
|
||||
return
|
||||
}
|
||||
state.currentPrice = currentPrice
|
||||
b.setState(tgID, state)
|
||||
}
|
||||
|
||||
// Determine condition automatically: above ask or below ask.
|
||||
var condition entities.AlertCondition
|
||||
if target.GreaterThanOrEqual(currentPrice.Ask) {
|
||||
condition = entities.AlertConditionAbove
|
||||
} else {
|
||||
condition = entities.AlertConditionBelow
|
||||
state.targetPrice = target
|
||||
state.step = stepAddAlertAwaitType
|
||||
b.setState(tgID, state)
|
||||
|
||||
kb := tgbotapi.NewInlineKeyboardMarkup(
|
||||
tgbotapi.NewInlineKeyboardRow(
|
||||
tgbotapi.NewInlineKeyboardButtonData("Price crossing", "alert_type:crossing"),
|
||||
tgbotapi.NewInlineKeyboardButtonData("Candle close", "alert_type:close"),
|
||||
),
|
||||
)
|
||||
msg := tgbotapi.NewMessage(chatID, fmt.Sprintf("Target price: %s\n\nSelect alert type:", target.String()))
|
||||
msg.ReplyMarkup = kb
|
||||
b.sendMsg(msg)
|
||||
}
|
||||
|
||||
// handleAlertTypeCrossing creates a standard crossing alert (High/Low vs target).
|
||||
func (b *Bot) handleAlertTypeCrossing(ctx context.Context, tgID entities.TelegramID, chatID int64) {
|
||||
state := b.getState(tgID)
|
||||
if state.step != stepAddAlertAwaitType {
|
||||
return
|
||||
}
|
||||
|
||||
condition := b.detectCondition(state)
|
||||
|
||||
user, err := b.requireUser(ctx, tgID, chatID)
|
||||
if err != nil {
|
||||
return
|
||||
|
|
@ -544,7 +615,7 @@ func (b *Bot) handleAddAlertPrice(ctx context.Context, tgID entities.TelegramID,
|
|||
|
||||
alert := &entities.Alert{
|
||||
UserID: user.ID,
|
||||
Price: target,
|
||||
Price: state.targetPrice,
|
||||
Condition: condition,
|
||||
Instrument: state.instrument,
|
||||
}
|
||||
|
|
@ -561,12 +632,98 @@ func (b *Bot) handleAddAlertPrice(ctx context.Context, tgID entities.TelegramID,
|
|||
b.alerter.AddAlert(alert)
|
||||
b.setState(tgID, &userState{})
|
||||
b.sendMenu(chatID, fmt.Sprintf(
|
||||
"Alert created!\n\n%s/%s %s %s\n\nYou will be notified when the price reaches your target.",
|
||||
"Alert created!\n\n%s/%s %s %s\n\nYou will be notified when the price crosses your target.",
|
||||
state.instrument.BaseCurrency, state.instrument.QuoteCurrency,
|
||||
condition, target.String(),
|
||||
formatCondition(condition), state.targetPrice.String(),
|
||||
))
|
||||
}
|
||||
|
||||
// handleAlertTypeClose shows a timeframe selection keyboard for candle-close alerts.
|
||||
func (b *Bot) handleAlertTypeClose(tgID entities.TelegramID, chatID int64) {
|
||||
state := b.getState(tgID)
|
||||
if state.step != stepAddAlertAwaitType {
|
||||
return
|
||||
}
|
||||
|
||||
// Build one row of timeframe buttons.
|
||||
var row []tgbotapi.InlineKeyboardButton
|
||||
for _, tf := range offeredTimeframes {
|
||||
row = append(row, tgbotapi.NewInlineKeyboardButtonData(
|
||||
tf.label,
|
||||
fmt.Sprintf("alert_timeframe:%s", tf.interval),
|
||||
))
|
||||
}
|
||||
|
||||
msg := tgbotapi.NewMessage(chatID, "Select the candle timeframe:")
|
||||
msg.ReplyMarkup = tgbotapi.NewInlineKeyboardMarkup(row)
|
||||
b.sendMsg(msg)
|
||||
}
|
||||
|
||||
// handleAlertTimeframe creates a candle-close alert for the selected timeframe.
|
||||
func (b *Bot) handleAlertTimeframe(ctx context.Context, tgID entities.TelegramID, chatID int64, tf provider.KlineInterval) {
|
||||
state := b.getState(tgID)
|
||||
if state.step != stepAddAlertAwaitType {
|
||||
return
|
||||
}
|
||||
|
||||
condition := b.detectCloseCondition(state)
|
||||
|
||||
user, err := b.requireUser(ctx, tgID, chatID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
alert := &entities.Alert{
|
||||
UserID: user.ID,
|
||||
Price: state.targetPrice,
|
||||
Condition: condition,
|
||||
Instrument: state.instrument,
|
||||
Timeframe: string(tf),
|
||||
}
|
||||
|
||||
id, err := b.usecase.CreateAlert(ctx, alert)
|
||||
if err != nil {
|
||||
b.log.Error("failed to create alert", "err", err)
|
||||
b.sendMenu(chatID, "Failed to create alert. Please try again.")
|
||||
b.setState(tgID, &userState{})
|
||||
return
|
||||
}
|
||||
|
||||
alert.ID = id
|
||||
b.alerter.AddAlert(alert)
|
||||
b.setState(tgID, &userState{})
|
||||
b.sendMenu(chatID, fmt.Sprintf(
|
||||
"Alert created!\n\n%s/%s %s %s (%s)\n\nYou will be notified when a %s candle closes %s the target.",
|
||||
state.instrument.BaseCurrency, state.instrument.QuoteCurrency,
|
||||
formatCondition(condition), state.targetPrice.String(), string(tf),
|
||||
string(tf),
|
||||
closeDirectionWord(condition),
|
||||
))
|
||||
}
|
||||
|
||||
// detectCondition returns above/below based on target vs current ask.
|
||||
func (b *Bot) detectCondition(state *userState) entities.AlertCondition {
|
||||
if state.currentPrice != nil && state.targetPrice.GreaterThanOrEqual(state.currentPrice.Ask) {
|
||||
return entities.AlertConditionAbove
|
||||
}
|
||||
return entities.AlertConditionBelow
|
||||
}
|
||||
|
||||
// detectCloseCondition returns close_above/close_below based on target vs current ask.
|
||||
func (b *Bot) detectCloseCondition(state *userState) entities.AlertCondition {
|
||||
if state.currentPrice != nil && state.targetPrice.GreaterThanOrEqual(state.currentPrice.Ask) {
|
||||
return entities.AlertConditionCloseAbove
|
||||
}
|
||||
return entities.AlertConditionCloseBelow
|
||||
}
|
||||
|
||||
func closeDirectionWord(c entities.AlertCondition) string {
|
||||
if c == entities.AlertConditionCloseAbove {
|
||||
return "above"
|
||||
}
|
||||
return "below"
|
||||
}
|
||||
|
||||
// handleEditAlertStart begins the edit flow for a specific alert.
|
||||
func (b *Bot) handleEditAlertStart(tgID entities.TelegramID, chatID int64, alertID entities.AlertID) {
|
||||
b.setState(tgID, &userState{
|
||||
|
|
|
|||
|
|
@ -7,8 +7,10 @@ type AlertID string
|
|||
type AlertCondition string
|
||||
|
||||
const (
|
||||
AlertConditionAbove AlertCondition = "above" // trigger when price rises to target
|
||||
AlertConditionBelow AlertCondition = "below" // trigger when price drops to target
|
||||
AlertConditionAbove AlertCondition = "above" // trigger when candle High reaches target
|
||||
AlertConditionBelow AlertCondition = "below" // trigger when candle Low reaches target
|
||||
AlertConditionCloseAbove AlertCondition = "close_above" // trigger when candle Close exceeds target
|
||||
AlertConditionCloseBelow AlertCondition = "close_below" // trigger when candle Close drops below target
|
||||
)
|
||||
|
||||
type Alert struct {
|
||||
|
|
@ -17,4 +19,5 @@ type Alert struct {
|
|||
Price decimal.Decimal
|
||||
Condition AlertCondition
|
||||
Instrument Instrument
|
||||
Timeframe string // non-empty for close_above / close_below; provider.KlineInterval value
|
||||
}
|
||||
|
|
|
|||
|
|
@ -10,4 +10,5 @@ type Candle struct {
|
|||
OpenTime time.Time
|
||||
High decimal.Decimal
|
||||
Low decimal.Decimal
|
||||
Close decimal.Decimal
|
||||
}
|
||||
|
|
|
|||
|
|
@ -135,8 +135,8 @@ func (b *Bybit) Candles(ctx context.Context, instrument entities.Instrument, fro
|
|||
}
|
||||
|
||||
for _, item := range resp.List {
|
||||
if len(item) < 4 {
|
||||
b.log.Error("bybit candles: length of elements less then 4", "len", len(item))
|
||||
if len(item) < 5 {
|
||||
b.log.Error("bybit candles: length of elements less than 5", "len", len(item))
|
||||
continue
|
||||
}
|
||||
startMs, err := strconv.ParseInt(item[0], 10, 64)
|
||||
|
|
@ -154,10 +154,16 @@ func (b *Bybit) Candles(ctx context.Context, instrument entities.Instrument, fro
|
|||
b.log.Error("bybit candles: failed to parse candle low price", "err", err)
|
||||
continue
|
||||
}
|
||||
close, err := decimal.NewFromString(item[4])
|
||||
if err != nil {
|
||||
b.log.Error("bybit candles: failed to parse candle close price", "err", err)
|
||||
continue
|
||||
}
|
||||
allCandles = append(allCandles, entities.Candle{
|
||||
OpenTime: time.UnixMilli(startMs),
|
||||
High: high,
|
||||
Low: low,
|
||||
Close: close,
|
||||
})
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -9,14 +9,19 @@ import (
|
|||
)
|
||||
|
||||
const saveAlertQuery = `
|
||||
insert into alert(user_id, instrument_id, price, condition)
|
||||
values ($1, $2, $3, $4)
|
||||
insert into alert(user_id, instrument_id, price, condition, timeframe)
|
||||
values ($1, $2, $3, $4, $5)
|
||||
returning id`
|
||||
|
||||
func (p *Postgresql) SaveAlert(ctx context.Context, alert *entities.Alert) (entities.AlertID, error) {
|
||||
var id entities.AlertID
|
||||
|
||||
err := p.db.QueryRow(ctx, saveAlertQuery, alert.UserID, alert.Instrument.ID, alert.Price.String(), alert.Condition).Scan(&id)
|
||||
var timeframe *string
|
||||
if alert.Timeframe != "" {
|
||||
timeframe = &alert.Timeframe
|
||||
}
|
||||
|
||||
err := p.db.QueryRow(ctx, saveAlertQuery, alert.UserID, alert.Instrument.ID, alert.Price.String(), alert.Condition, timeframe).Scan(&id)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to exec saveAlertQuery: %w", err)
|
||||
}
|
||||
|
|
@ -25,13 +30,13 @@ func (p *Postgresql) SaveAlert(ctx context.Context, alert *entities.Alert) (enti
|
|||
}
|
||||
|
||||
const allActiveAlertsQuery = `
|
||||
select a.id, a.user_id, a.price, a.condition, i.id, c_base.symbol, c_quote.symbol
|
||||
select a.id, a.user_id, a.price, a.condition, a.timeframe, i.id, c_base.symbol, c_quote.symbol
|
||||
from alert a
|
||||
join instrument i on i.id = a.instrument_id
|
||||
join currency c_base on c_base.id = i.base_currency_id
|
||||
join currency c_quote on c_quote.id = i.quoted_currency_id
|
||||
where a.active = true
|
||||
order by a.id`
|
||||
order by a.created_at desc`
|
||||
|
||||
func (p *Postgresql) AllActiveAlerts(ctx context.Context) ([]entities.Alert, error) {
|
||||
rows, err := p.db.Query(ctx, allActiveAlertsQuery)
|
||||
|
|
@ -44,9 +49,10 @@ func (p *Postgresql) AllActiveAlerts(ctx context.Context) ([]entities.Alert, err
|
|||
for rows.Next() {
|
||||
var alert entities.Alert
|
||||
var priceStr string
|
||||
var timeframe *string
|
||||
|
||||
if err := rows.Scan(
|
||||
&alert.ID, &alert.UserID, &priceStr, &alert.Condition,
|
||||
&alert.ID, &alert.UserID, &priceStr, &alert.Condition, &timeframe,
|
||||
&alert.Instrument.ID, &alert.Instrument.BaseCurrency, &alert.Instrument.QuoteCurrency,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("failed to scan alert row: %w", err)
|
||||
|
|
@ -57,6 +63,10 @@ func (p *Postgresql) AllActiveAlerts(ctx context.Context) ([]entities.Alert, err
|
|||
return nil, fmt.Errorf("failed to parse alert price: %w", err)
|
||||
}
|
||||
|
||||
if timeframe != nil {
|
||||
alert.Timeframe = *timeframe
|
||||
}
|
||||
|
||||
alerts = append(alerts, alert)
|
||||
}
|
||||
|
||||
|
|
@ -64,7 +74,7 @@ func (p *Postgresql) AllActiveAlerts(ctx context.Context) ([]entities.Alert, err
|
|||
}
|
||||
|
||||
const alertByIDQuery = `
|
||||
select a.id, a.user_id, a.price, a.condition, i.id, c_base.symbol, c_quote.symbol
|
||||
select a.id, a.user_id, a.price, a.condition, a.timeframe, i.id, c_base.symbol, c_quote.symbol
|
||||
from alert a
|
||||
join instrument i on i.id = a.instrument_id
|
||||
join currency c_base on c_base.id = i.base_currency_id
|
||||
|
|
@ -74,9 +84,10 @@ where a.id = $1 and a.active = true`
|
|||
func (p *Postgresql) AlertByID(ctx context.Context, id entities.AlertID) (*entities.Alert, error) {
|
||||
var alert entities.Alert
|
||||
var priceStr string
|
||||
var timeframe *string
|
||||
|
||||
err := p.db.QueryRow(ctx, alertByIDQuery, id).Scan(
|
||||
&alert.ID, &alert.UserID, &priceStr, &alert.Condition,
|
||||
&alert.ID, &alert.UserID, &priceStr, &alert.Condition, &timeframe,
|
||||
&alert.Instrument.ID, &alert.Instrument.BaseCurrency, &alert.Instrument.QuoteCurrency,
|
||||
)
|
||||
if err != nil {
|
||||
|
|
@ -88,17 +99,21 @@ func (p *Postgresql) AlertByID(ctx context.Context, id entities.AlertID) (*entit
|
|||
return nil, fmt.Errorf("failed to parse alert price: %w", err)
|
||||
}
|
||||
|
||||
if timeframe != nil {
|
||||
alert.Timeframe = *timeframe
|
||||
}
|
||||
|
||||
return &alert, nil
|
||||
}
|
||||
|
||||
const alertsByUserIDQuery = `
|
||||
select a.id, a.user_id, a.price, a.condition, i.id, c_base.symbol, c_quote.symbol
|
||||
select a.id, a.user_id, a.price, a.condition, a.timeframe, i.id, c_base.symbol, c_quote.symbol
|
||||
from alert a
|
||||
join instrument i on i.id = a.instrument_id
|
||||
join currency c_base on c_base.id = i.base_currency_id
|
||||
join currency c_quote on c_quote.id = i.quoted_currency_id
|
||||
where a.user_id = $1 and a.active = true
|
||||
order by a.id
|
||||
order by a.created_at desc
|
||||
offset $2 limit $3`
|
||||
|
||||
func (p *Postgresql) AlertsByUserID(ctx context.Context, userID entities.UserID, offset, limit int) ([]entities.Alert, error) {
|
||||
|
|
@ -112,9 +127,10 @@ func (p *Postgresql) AlertsByUserID(ctx context.Context, userID entities.UserID,
|
|||
for rows.Next() {
|
||||
var alert entities.Alert
|
||||
var priceStr string
|
||||
var timeframe *string
|
||||
|
||||
if err := rows.Scan(
|
||||
&alert.ID, &alert.UserID, &priceStr, &alert.Condition,
|
||||
&alert.ID, &alert.UserID, &priceStr, &alert.Condition, &timeframe,
|
||||
&alert.Instrument.ID, &alert.Instrument.BaseCurrency, &alert.Instrument.QuoteCurrency,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("failed to scan alert row: %w", err)
|
||||
|
|
@ -125,6 +141,10 @@ func (p *Postgresql) AlertsByUserID(ctx context.Context, userID entities.UserID,
|
|||
return nil, fmt.Errorf("failed to parse alert price: %w", err)
|
||||
}
|
||||
|
||||
if timeframe != nil {
|
||||
alert.Timeframe = *timeframe
|
||||
}
|
||||
|
||||
alerts = append(alerts, alert)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,7 @@
|
|||
alter table alert drop column timeframe;
|
||||
|
||||
-- PostgreSQL does not support removing enum values; recreate the type without close_above/close_below.
|
||||
alter table alert alter column condition type text;
|
||||
drop type alert_condition;
|
||||
create type alert_condition as enum ('above', 'below');
|
||||
alter table alert alter column condition type alert_condition using condition::alert_condition;
|
||||
|
|
@ -0,0 +1,4 @@
|
|||
alter type alert_condition add value 'close_above';
|
||||
alter type alert_condition add value 'close_below';
|
||||
|
||||
alter table alert add column timeframe text;
|
||||
|
|
@ -0,0 +1 @@
|
|||
alter table alert drop column created_at;
|
||||
|
|
@ -0,0 +1 @@
|
|||
alter table alert add column created_at timestamptz not null default now();
|
||||
|
|
@ -133,36 +133,76 @@ func selectCandleInterval(gap time.Duration) provider.KlineInterval {
|
|||
}
|
||||
|
||||
// TODO: parallel checking for different instruments.
|
||||
// TODO: get one candle before interval
|
||||
|
||||
func (a *Alerter) checkAlerts(ctx context.Context) error {
|
||||
now := time.Now()
|
||||
gap := now.Sub(a.lastCheckedAt)
|
||||
candleInterval := selectCandleInterval(gap)
|
||||
|
||||
// Truncate to the selected interval boundary so we always re-check the candle
|
||||
// that was still forming at the previous tick — its High/Low may have changed
|
||||
// since.
|
||||
from := a.lastCheckedAt.Truncate(candleInterval.ToDuration())
|
||||
|
||||
a.log.Debug("checking alerts", "from", from, "to", now, "interval", candleInterval, "gap", gap.Round(time.Second))
|
||||
|
||||
instruments := a.cache.Instruments()
|
||||
|
||||
for _, instrument := range instruments {
|
||||
alerts := a.cache.AlertsByInstrument(instrument.ID)
|
||||
|
||||
// Separate crossing alerts (above/below) from candle-close alerts.
|
||||
var crossingAlerts []*entities.Alert
|
||||
closeByTimeframe := make(map[provider.KlineInterval][]*entities.Alert)
|
||||
|
||||
for _, alert := range alerts {
|
||||
switch alert.Condition {
|
||||
case entities.AlertConditionAbove, entities.AlertConditionBelow:
|
||||
crossingAlerts = append(crossingAlerts, alert)
|
||||
case entities.AlertConditionCloseAbove, entities.AlertConditionCloseBelow:
|
||||
tf := provider.KlineInterval(alert.Timeframe)
|
||||
closeByTimeframe[tf] = append(closeByTimeframe[tf], alert)
|
||||
}
|
||||
}
|
||||
|
||||
// Check crossing alerts using an auto-selected interval for the gap.
|
||||
if len(crossingAlerts) > 0 {
|
||||
gap := now.Sub(a.lastCheckedAt)
|
||||
candleInterval := selectCandleInterval(gap)
|
||||
from := a.lastCheckedAt.Truncate(candleInterval.ToDuration())
|
||||
|
||||
a.log.Debug("checking crossing alerts", "instrument", instrument.ID,
|
||||
"from", from, "to", now, "interval", candleInterval)
|
||||
|
||||
candles, err := a.priceProvider.Candles(ctx, instrument, from, now, candleInterval)
|
||||
if err != nil {
|
||||
a.log.Error("failed to get candles", "instrument", instrument.ID, "err", err)
|
||||
a.log.Error("failed to get candles for crossing alerts", "instrument", instrument.ID, "err", err)
|
||||
return fmt.Errorf("failed to get candles: %w", err)
|
||||
}
|
||||
|
||||
for _, alert := range a.cache.AlertsByInstrument(instrument.ID) {
|
||||
if triggered, price := alertTriggeredByCandles(alert, candles); triggered {
|
||||
for _, alert := range crossingAlerts {
|
||||
if triggered, price := crossingTriggered(alert, candles); triggered {
|
||||
a.triggerAlert(ctx, alert, price)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Check candle-close alerts per timeframe; only closed candles count.
|
||||
for tf, tfAlerts := range closeByTimeframe {
|
||||
from := a.lastCheckedAt.Truncate(tf.ToDuration())
|
||||
|
||||
a.log.Debug("checking close alerts", "instrument", instrument.ID,
|
||||
"from", from, "to", now, "timeframe", tf)
|
||||
|
||||
candles, err := a.priceProvider.Candles(ctx, instrument, from, now, tf)
|
||||
if err != nil {
|
||||
a.log.Error("failed to get candles for close alerts", "instrument", instrument.ID,
|
||||
"timeframe", tf, "err", err)
|
||||
return fmt.Errorf("failed to get candles: %w", err)
|
||||
}
|
||||
|
||||
// Only consider candles that have fully closed.
|
||||
closed := closedCandles(candles, tf, now)
|
||||
|
||||
for _, alert := range tfAlerts {
|
||||
if triggered, price := closeTriggered(alert, closed); triggered {
|
||||
a.triggerAlert(ctx, alert, price)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
a.lastCheckedAt = now
|
||||
|
||||
if err := a.storage.SetLastAlertCheck(ctx, now); err != nil {
|
||||
|
|
@ -172,9 +212,20 @@ func (a *Alerter) checkAlerts(ctx context.Context) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// alertTriggeredByCandles returns true and the triggering price if any candle
|
||||
// caused the alert condition to be met.
|
||||
func alertTriggeredByCandles(alert *entities.Alert, candles []entities.Candle) (bool, decimal.Decimal) {
|
||||
// closedCandles filters to candles whose close time (openTime + interval) is before now.
|
||||
func closedCandles(candles []entities.Candle, interval provider.KlineInterval, now time.Time) []entities.Candle {
|
||||
dur := interval.ToDuration()
|
||||
var result []entities.Candle
|
||||
for _, c := range candles {
|
||||
if !c.OpenTime.Add(dur).After(now) {
|
||||
result = append(result, c)
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// crossingTriggered returns true if any candle's High/Low crosses the alert price.
|
||||
func crossingTriggered(alert *entities.Alert, candles []entities.Candle) (bool, decimal.Decimal) {
|
||||
for _, candle := range candles {
|
||||
switch alert.Condition {
|
||||
case entities.AlertConditionAbove:
|
||||
|
|
@ -190,6 +241,23 @@ func alertTriggeredByCandles(alert *entities.Alert, candles []entities.Candle) (
|
|||
return false, decimal.Zero
|
||||
}
|
||||
|
||||
// closeTriggered returns true if any closed candle's Close crosses the alert price.
|
||||
func closeTriggered(alert *entities.Alert, candles []entities.Candle) (bool, decimal.Decimal) {
|
||||
for _, candle := range candles {
|
||||
switch alert.Condition {
|
||||
case entities.AlertConditionCloseAbove:
|
||||
if candle.Close.GreaterThanOrEqual(alert.Price) {
|
||||
return true, candle.Close
|
||||
}
|
||||
case entities.AlertConditionCloseBelow:
|
||||
if candle.Close.LessThanOrEqual(alert.Price) {
|
||||
return true, candle.Close
|
||||
}
|
||||
}
|
||||
}
|
||||
return false, decimal.Zero
|
||||
}
|
||||
|
||||
func (a *Alerter) triggerAlert(ctx context.Context, alert *entities.Alert, currentPrice decimal.Decimal) {
|
||||
if err := a.notifier.NotifyAlert(ctx, alert.UserID, alert, currentPrice); err != nil {
|
||||
a.log.Error("failed to notify alert", "alert_id", alert.ID, "err", err)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue