alerts

package module
v0.6.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 16, 2026 License: MIT Imports: 30 Imported by: 0

README

kite-mcp-alerts

Go Reference

Alert engine + briefing scheduler for the algo2go ecosystem. Provides price-threshold alerts, composite alert chains, trailing-stop alerts, anomaly notifier, morning/EOD briefings, and SQLite-backed alert store with credential + token persistence.

Used by Sundeepg98/kite-mcp-server for the alerts MCP toolset, scheduler-driven dispatch, telegram notifier, and audit-log integration.

Why a separate module?

Alert evaluation is a foundational primitive that other algo2go projects (broker dashboards, monitoring, future trading bots) need independent of kite-mcp-server. Hosting as a module:

  • Centralizes the alert engine + briefing scheduler across consumers
  • Enables independent versioning of alert types + composite logic
  • Keeps the dep-graph weight focused on consumers that actually need alert evaluation

Stability promise

v0.x — unstable. Type signatures may evolve as alert logic matures. Pin v0.1.0 deliberately. v1.0 ships only after the public API (Store, BriefingService, AnomalyNotifier, evaluator types) is reviewed for stability and at least one external consumer ships against it.

Install

go get github.com/algo2go/kite-mcp-alerts@v0.1.0

Public API (selected)

Store + persistence
  • Store — alert + composite + trailing CRUD with SQLite backend
  • DB — direct DB adapter for tokens, sessions, telegram bindings
  • NewStore(notifyCallback) — store constructor
Briefings
  • BriefingService — IST-aligned morning/EOD dispatcher
  • KiteAdapter integration via BrokerDataProvider interface
Alerts
  • Evaluator — price + composite + trailing alert evaluation
  • AnomalyNotifier — μ+3σ anomaly alerts via Telegram
Telegram
  • TelegramSender — bot wrapper for alert dispatch

Dependencies

  • github.com/algo2go/kite-mcp-broker — KiteSDK + zerodha adapter
  • github.com/algo2go/kite-mcp-domain — Alert + CompositeAlert entities
  • github.com/algo2go/kite-mcp-isttz — IST market hours
  • github.com/algo2go/kite-mcp-logger — structured logging
  • github.com/algo2go/kite-mcp-money (indirect via kc/domain)
  • github.com/zerodha/gokiteconnect/v4 — Kite SDK
  • github.com/go-telegram-bot-api/telegram-bot-api/v5 — Telegram
  • modernc.org/sqlite — SQLite driver

All algo2go deps are published modules; no upstream replace directives needed.

Test surface

Most tests run standalone (~30+ test functions). One test file (briefing_injection_test.go) was stripped during extraction because it imported the unpublished testutil module from kite-mcp-server's workspace. That test still runs in the consumer's workspace mode where testutil resolves in-tree.

Reference consumer

Sundeepg98/kite-mcp-server — consumed by:

  • kc/alert_service.go, kc/manager_init.go — service wiring
  • kc/usecases/alert_usecases.go, create_alert.go, etc. — use cases
  • kc/audit/store.go — audit projection
  • kc/billing/store.go — billing event integration
  • mcp/alerts/*.go — MCP tools (set_alert, list_alerts, etc.)
  • kc/papertrading, kc/telegram — paper-trading + Telegram integration

License

MIT — see LICENSE.

Authors

Original design + alert engine: Sundeepg98 (Zerodha Tech). Multi-module promotion (2026-05-10): algo2go contributors.

Documentation

Index

Constants

View Source 
const (
	DirectionAbove   = domain.DirectionAbove
	DirectionBelow   = domain.DirectionBelow
	DirectionDropPct = domain.DirectionDropPct
	DirectionRisePct = domain.DirectionRisePct

	// MaxAlertsPerUser is the maximum number of alerts a single user can have.
	// Persistence-layer limit, stays in this package.
	MaxAlertsPerUser = 100
)

Alert direction constants. Re-exported from kc/domain for call-site compatibility — alerts.DirectionAbove still works.

View Source 
const (
	CompositeLogicAnd  = domain.CompositeLogicAnd
	CompositeLogicAny  = domain.CompositeLogicAny
	AlertTypeSingle    = domain.AlertTypeSingle
	AlertTypeComposite = domain.AlertTypeComposite
)

Composite logic constants — re-exported so alerts.CompositeLogicAnd etc. resolve without the caller needing to import kc/domain.

Variables

View Source 
var ValidDirections = domain.ValidDirections

ValidDirections re-exports the domain's validation set.

Functions

func ColumnExists added in v0.3.0 

func ColumnExists(d Dialect, db *sql.DB, table, column string) (bool, error)

ColumnExists reports whether the named column is present on the given table. Used for incremental schema migrations that ALTER TABLE ADD COLUMN only when the column is absent. 

SQLite:   SELECT COUNT(*) FROM pragma_table_info(<table>) WHERE name=?
          (table name is positional, not parametric, in pragma syntax)
Postgres: SELECT 1 FROM information_schema.columns
          WHERE table_schema=current_schema()
            AND table_name=$1 AND column_name=$2

Returns (false, nil) iff the column is absent. Returns (false, err) only on database errors. The SQLite branch silently returns (false, nil) when the parent table is absent (matching the legacy pragma_table_info behavior).

SQLite naming caveat: pragma_table_info is a virtual table function that takes the target-table name as a string LITERAL, not a bind parameter. We sanity-check the name to permit only [a-zA-Z0-9_] to prevent SQL injection through the literal-substitution path.

Callers: db_migrations.go's migrateAlerts (replaces the direct pragma_table_info query at v0.2.0); db.go's mcp_sessions.session_id_enc migration; kite-mcp-billing's admin_email-column check.

func Decrypt 

func Decrypt(key []byte, hexCiphertext string) string

Decrypt decrypts hex-encoded AES-256-GCM ciphertext with the given key. This is the exported wrapper around decrypt for use by external tools (e.g. key rotation CLI).

func DeriveEncryptionKey 

func DeriveEncryptionKey(secret string) ([]byte, error)

DeriveEncryptionKey derives a 32-byte AES-256 key from a secret using HKDF-SHA256 with a nil salt (legacy). Retained for backward compatibility during migration. New code should use DeriveEncryptionKeyWithSalt.

func DeriveEncryptionKeyWithSalt 

func DeriveEncryptionKeyWithSalt(secret string, salt []byte) ([]byte, error)

DeriveEncryptionKeyWithSalt derives a 32-byte AES-256 key from a secret using HKDF-SHA256 with the provided salt. Per RFC 5869, a random salt strengthens the extract step and is recommended over a nil/zero salt.

func Encrypt 

func Encrypt(key []byte, plaintext string) (string, error)

Encrypt encrypts plaintext using AES-256-GCM with the given key. This is the exported wrapper around encrypt for use by external tools (e.g. key rotation CLI).

func EnsureEncryptionSalt 

func EnsureEncryptionSalt(db *DB, secret string) ([]byte, error)

EnsureEncryptionSalt ensures a random 32-byte HKDF salt exists in the database, derives the salted encryption key, and migrates all encrypted data from the old nil-salt key to the new salted key on first run. On subsequent runs it simply loads the salt and returns the salted key.

Returns the salted encryption key ready for use.

func EscapeMarkdown 

func EscapeMarkdown(s string) string

EscapeMarkdown escapes special MarkdownV2 characters for Telegram messages. This is the exported version of escapeTelegramMarkdown.

func IsPercentageDirection 

func IsPercentageDirection(d Direction) bool

IsPercentageDirection re-exports the domain helper.

func OverrideNewBotFunc 

func OverrideNewBotFunc(fn func(string) (BotAPI, error)) func()

OverrideNewBotFunc replaces the bot creation function and returns a cleanup function that restores the original. Safe for concurrent tests.

func PragmaInit added in v0.3.0 

func PragmaInit(d Dialect, db *sql.DB) error

PragmaInit applies dialect-specific connection-init pragmas. 

SQLite:   PRAGMA journal_mode=WAL + PRAGMA busy_timeout=5000.
          foreign_keys=ON is handled at the DSN layer
          (dsnWithFKPragma) per-connection because pool reopens
          reset the per-connection PRAGMA state.
Postgres: no-op. WAL is the only journal mode (cluster-level);
          statement_timeout / lock_timeout are configured in
          postgresql.conf or via SET on the session; FK
          enforcement is always ON and cannot be disabled per
          connection.

Returns nil iff the database accepted all required init pragmas (or, for Postgres, immediately).

Callers: OpenDB (db.go) for SQLite. Postgres adapter at Phase 2.2 passes its newly-opened *sql.DB through PragmaInit(DialectPostgres) for symmetric init flow even though the body is empty.

func RestoreNewBotFunc deprecated 

func RestoreNewBotFunc()

RestoreNewBotFunc is a no-op placeholder: the original factory cannot be captured generically across tests. Instead, use OverrideNewBotFunc which returns a restore function.

Deprecated: Use OverrideNewBotFunc instead.

func SchemaDDL added in v0.3.0 

func SchemaDDL(d Dialect) string

SchemaDDL returns the dialect-specific CREATE TABLE / CREATE INDEX block for the kite-mcp-alerts shared schema (alerts, telegram_chat_ids, kite_tokens, kite_credentials, oauth_clients, mcp_sessions, config, trailing_stops, daily_pnl, app_registry).

Phase 2.2: returns truly dialect-flavored DDL.

  • DialectSQLite → sqliteSchemaDDL (legacy form preserved for byte-identical compatibility with OpenDB's inline DDL literal; OpenDB still uses its inline copy at v0.4.0)
  • DialectPostgres → postgresSchemaDDL (REAL → DOUBLE PRECISION, INTEGER instrument_token → BIGINT, INTEGER boolean columns kept as INTEGER for dump/load portability)

The Postgres mapping deliberately preserves INTEGER (not BOOLEAN) for the 0/1-encoded boolean columns (triggered, terminated, active, is_kite_key) so that cross-dialect data export/import does NOT require value transformation. Storage cost identical (4 bytes); the columns are read with `triggeredI != 0` semantics in Go anyway.

REAL → DOUBLE PRECISION is the only mandatory cosmetic substitution — Postgres parses REAL as 4-byte float (32-bit), but our prices are float64. DOUBLE PRECISION is the explicit 8-byte form.

instrument_token INTEGER → BIGINT: NSE/BSE instrument tokens can exceed int32 range (32-bit signed: 2.1B max). BIGINT is 8-byte signed (9.2 quintillion). SQLite's INTEGER is variable-width and always supports int64; Postgres INTEGER is fixed 4-byte.

func SetNewBotFuncForTest 

func SetNewBotFuncForTest(fn func(string) (BotAPI, error))

SetNewBotFuncForTest replaces the bot creation function for cross-package testing. Callers must defer RestoreNewBotFunc to avoid leaking state.

func TableExists added in v0.3.0 

func TableExists(d Dialect, db *sql.DB, name string) (bool, error)

TableExists reports whether a table with the given name is present in the connected database. Uses the dialect's catalog query: 

SQLite:   SELECT 1 FROM sqlite_master WHERE type='table' AND name=?
Postgres: SELECT 1 FROM information_schema.tables
          WHERE table_schema=current_schema() AND table_name=$1

Returns (false, nil) iff the table is absent. Returns (false, err) only on database errors; a missing table is NOT an error.

Callers: db_migrations.go's migrateRegistryCheckConstraint (replaces the direct sqlite_master query at v0.2.0); Postgres-readiness for kite-mcp-billing's legacy self-migration (Phase 2.1.6 cross-repo edit at billing/store.go:142+161).

Types

type Alert 

type Alert = domain.Alert

Alert is an alias for domain.Alert — the canonical entity definition.

type AlertStoreInterface 

type AlertStoreInterface interface {
	// Add creates a new alert and returns its ID.
	Add(email, tradingsymbol, exchange string, instrumentToken uint32, targetPrice float64, direction Direction) (string, error)

	// AddWithReferencePrice creates a new alert with an optional reference price.
	AddWithReferencePrice(email, tradingsymbol, exchange string, instrumentToken uint32, targetPrice float64, direction Direction, referencePrice float64) (string, error)

	// AddComposite creates a new composite alert combining multiple per-
	// instrument conditions via AND/ANY logic. Returns the alert ID.
	AddComposite(email, name string, logic CompositeLogic, conds []CompositeCondition) (string, error)

	// Delete removes an alert by ID for the given email.
	Delete(email, alertID string) error

	// DeleteByEmail removes all alerts for the given email.
	DeleteByEmail(email string)

	// List returns all alerts for the given email.
	List(email string) []*Alert

	// GetByToken returns all active (non-triggered) alerts matching the instrument token.
	GetByToken(instrumentToken uint32) []*Alert

	// MarkTriggered marks an alert as triggered with the current price.
	MarkTriggered(alertID string, currentPrice float64) bool

	// MarkNotificationSent records when a Telegram notification was sent.
	MarkNotificationSent(alertID string, sentAt time.Time)

	// ListAll returns all alerts grouped by email.
	ListAll() map[string][]*Alert

	// ActiveCount returns the number of active alerts for a user.
	ActiveCount(email string) int
}

AlertStoreInterface defines operations for managing per-user price alerts.

Anchor 5 PR 5.2 (per .research/anchor-5-prs-design.md): this interface was relocated from `kc/interfaces.go:30` to its owning package (kc/alerts) so that kc/ports/alert.go can eventually drop its kc-parent import in PR 5.3 (Wave B-2). The legacy `kc.AlertStoreInterface` is preserved as a single-line type alias in kc/interfaces.go to keep the existing 11+ reverse-dep call sites compiling unchanged. Both names reference the SAME interface — Go type aliases are not new types — so satisfaction by *alerts.Store is preserved at the alias site.

Telegram chat ID operations are separated into TelegramStoreInterface (still in kc/interfaces.go because TelegramStore types live there).

Method set (12 methods, identical to the pre-move interface — only the type qualifier changes since we are now in the alerts package): 

Add, AddWithReferencePrice, AddComposite — alert creation
Delete, DeleteByEmail                    — removal
List, ListAll                            — querying by email/all
GetByToken                               — querying by instrument token
MarkTriggered, MarkNotificationSent      — lifecycle transitions
ActiveCount                              — count for limit enforcement

type AlertType 

type AlertType = domain.AlertType

AlertType is a type alias for the domain alert type discriminator.

type AnomalyNotifier 

type AnomalyNotifier struct {
	// contains filtered or unexported fields
}

AnomalyNotifier subscribes to domain.RiskguardRejectionEvent and pushes a Telegram message to every configured admin's chat. It closes the PULL-only gap surfaced in observability-audit-and-roadmap.md §1.4 (the existing admin_list_anomaly_flags MCP tool requires the operator to remember to poll; this surface provides real-time push).

Per-(user_email, reason) deduplication with a 5-minute default TTL prevents alert storms when a single misconfigured user trips the same flag repeatedly. The dedup map is bounded by natural churn — entries time out without explicit cleanup goroutines.

Trader email is HASHED in the message body via SHA-256 (same shape as audit.HashEmail) — operators get a stable correlation key without PII exposure on the Telegram channel (which has its own infrastructure trust boundaries). The reason and timestamp ARE included plaintext; they're not user-identifying.

HandleEvent is the dispatcher Subscribe target. It MUST be quick (Dispatch holds an RLock); the actual Telegram send runs in a goroutine.

func NewAnomalyNotifier 

func NewAnomalyNotifier(notifier *TelegramNotifier, store *Store, admins []string, logger *slog.Logger) *AnomalyNotifier

NewAnomalyNotifier constructs an AnomalyNotifier with the default 5-minute dedup TTL. Pass nil notifier or empty admins for a no-op instance (HandleEvent silently skips when there are no recipients).

func (*AnomalyNotifier) HandleEvent 

func (a *AnomalyNotifier) HandleEvent(evt domain.Event)

HandleEvent processes a domain.RiskguardRejectionEvent (other event types are ignored — defensive against subscribe-misconfiguration).

Quick path: returns immediately after lookup-and-decision; the actual Telegram send runs in a goroutine to keep the dispatcher unblocked.

type BotAPI 

type BotAPI interface {
	Send(c tgbotapi.Chattable) (tgbotapi.Message, error)
	Request(c tgbotapi.Chattable) (*tgbotapi.APIResponse, error)
}

BotAPI abstracts the Telegram bot client for testability. *tgbotapi.BotAPI satisfies this interface.

type BotFactory 

type BotFactory func(token string) (BotAPI, error)

BotFactory constructs a BotAPI from a token. Used by NewTelegramNotifierWithFactory to inject per-call bot creation — the t.Parallel-safe alternative to the package-level newBotFunc global mutated via OverrideNewBotFunc.

type BriefingService 

type BriefingService struct {
	// contains filtered or unexported fields
}

BriefingService generates and sends morning briefings and daily P&L summaries via Telegram for users who have a registered chat ID and valid Kite token.

Wave D Phase 3 Package 4 (Logger sweep): logger is typed as the kc/logger.Logger port. NewBriefingService takes *slog.Logger for caller compatibility (app/wire.go's scheduler init) and converts at the boundary via logport.NewSlog. Internal log calls use context.Background() — Briefing is invoked from scheduler goroutines with no request ctx in scope.

func NewBriefingService 

func NewBriefingService(
	notifier *TelegramNotifier,
	alertStore *Store,
	tokens TokenChecker,
	creds CredentialGetter,
	logger *slog.Logger,
) *BriefingService

NewBriefingService creates a BriefingService. Returns nil if notifier is nil.

func (*BriefingService) SendDailySummaries 

func (b *BriefingService) SendDailySummaries()

SendDailySummaries sends a post-market P&L summary to every eligible user.

func (*BriefingService) SendMISWarnings 

func (b *BriefingService) SendMISWarnings()

SendMISWarnings sends a Telegram warning to users who have open MIS positions. Intended to run at 2:30 PM IST, giving ~45 minutes before auto square-off.

func (*BriefingService) SendMorningBriefings 

func (b *BriefingService) SendMorningBriefings()

SendMorningBriefings sends a morning briefing to every user with a Telegram chat ID.

func (*BriefingService) SetBrokerProvider 

func (b *BriefingService) SetBrokerProvider(p BrokerDataProvider)

SetBrokerProvider overrides the default Kite API client with a custom provider (for testing).

func (*BriefingService) SetKiteClientFactory 

func (b *BriefingService) SetKiteClientFactory(f KiteClientFactory)

SetKiteClientFactory wires the factory used by the default broker provider. Production wires this during app bootstrap; tests may leave it nil when they override the broker provider via SetBrokerProvider.

type BrokerDataProvider 

type BrokerDataProvider interface {
	// GetHoldings returns holdings for the user.
	GetHoldings(apiKey, accessToken string) ([]kiteconnect.Holding, error)
	// GetPositions returns positions for the user.
	GetPositions(apiKey, accessToken string) (kiteconnect.Positions, error)
	// GetUserMargins returns margin info for the user.
	GetUserMargins(apiKey, accessToken string) (kiteconnect.AllMargins, error)
	// GetLTP returns last traded prices for the given instruments.
	GetLTP(apiKey, accessToken string, instruments ...string) (kiteconnect.QuoteLTP, error)
}

BrokerDataProvider abstracts broker API calls for testability. When set on BriefingService via SetBrokerProvider, these methods are used instead of creating kiteconnect.Client directly.

F3 close-out (Phase B/D): the redundancy audit (commit 55ff79c) initially flagged this as a duplicate of broker.PortfolioReader. Empirical inspection shows the two interfaces are STRUCTURALLY DIFFERENT — they serve different architectural roles:

  • broker.PortfolioReader is the SESSION-PINNED port (3 methods: GetHoldings/GetPositions/GetTrades, no auth args, returns broker-agnostic DTOs). Implemented by *broker.Client; consumed by MCP tool handlers + use cases via the session-bound broker.

  • BrokerDataProvider is the FACTORY-BACKED port (4 methods: GetHoldings/GetPositions/GetUserMargins/GetLTP, ALL take (apiKey, accessToken) per-call, return RAW kiteconnect types). Used by background services (briefing scheduler, P&L snapshot scheduler) that run outside the MCP request lifecycle and therefore cannot reach a session-pinned broker — they construct a fresh kiteconnect client per email via KiteClientFactory.

Three concrete differences make merger structurally infeasible:

  1. Method signatures: per-call (apiKey, accessToken) vs none
  2. Return types: kiteconnect.* (raw SDK) vs broker.* (DTOs)
  3. Method sets: GetUserMargins+GetLTP here vs GetTrades there

Forcing alignment would either redesign broker.PortfolioReader's surface (forbidden per F3 brief stop condition) or require a translation layer at every BrokerDataProvider callsite (the very abstraction this interface exists to avoid). The names ARE similar enough to invite confusion; this godoc is the F3 resolution — document the divergence so future audits don't re-flag it.

If a third broker integration ever needs this background-service pattern, factor a shared port at that point. Today (Zerodha-only), keeping the local declaration matches the actual coupling.

type ClientDBEntry 

type ClientDBEntry struct {
	ClientID     string
	ClientSecret string
	RedirectURIs string // JSON-encoded []string
	ClientName   string
	CreatedAt    time.Time
	IsKiteAPIKey bool
}

ClientDBEntry represents an OAuth client stored in the database.

type CompositeCondition 

type CompositeCondition = domain.CompositeCondition

CompositeCondition is a type alias for the domain composite condition — keeps callers who import only kc/alerts working without pulling domain.

type CompositeLogic 

type CompositeLogic = domain.CompositeLogic

CompositeLogic is a type alias for the domain composite logic operator.

type CredentialEntry 

type CredentialEntry struct {
	Email     string
	APIKey    string
	APISecret string
	AppID     string
	StoredAt  time.Time
}

CredentialEntry represents a user's Kite developer app credentials stored in the database.

type CredentialGetter 

type CredentialGetter interface {
	// GetAPIKey returns the API key for the given email, falling back to global.
	GetAPIKey(email string) string
}

CredentialGetter abstracts looking up per-user Kite credentials.

type DB 

type DB struct {
	// contains filtered or unexported fields
}

DB provides persistence for alerts and Telegram chat IDs.

Phase 2.2: now backed by either SQLite (modernc.org/sqlite) or Postgres (jackc/pgx/v5/stdlib). The driver field tags the open connection's dialect so dialect.go helpers can dispatch correctly from a *DB receiver. OpenDB → DialectSQLite; OpenPostgresDB → DialectPostgres.

func OpenDB 

func OpenDB(path string) (*DB, error)

OpenDB opens (or creates) the SQLite database at path and ensures tables exist.

func OpenLibSQL added in v0.6.0 

func OpenLibSQL(url string) (*DB, error)

OpenLibSQL opens (or connects to) a Turso/libSQL database at the given libsql:// URL and ensures schema exists.

Phase 2.6 Path 6 deliverable: makes Turso/libSQL a third backend option alongside SQLite and Postgres. The returned *DB is fully interchangeable with the SQLite *DB at the SQLDB interface level — libSQL is SQLite-compatible at the protocol layer, including `?` placeholders, ON CONFLICT, sqlite_master, pragma_table_info.

URL format: 

libsql://<dbname>-<username>.<region>.turso.io?authToken=<JWT>

Example: 

libsql://phase-2-6-canary-sundeepg98.aws-ap-south-1.turso.io?authToken=eyJ...

Auth token is supplied as a query-string param per the libsql- client-go driver convention. URL-only constructor by design (vs separate URL+token args) so the env-var binding is single-string (ALERT_DB_URL) — same shape as Postgres adapter.

Connection pool: libsql-client-go registers as the "libsql" sql.Driver; uses Go's default pool settings. Turso server-side handles concurrency; client-side single-pool is fine.

Schema bootstrap: applies SchemaDDL(DialectLibSQL) at open time — which returns the SQLite-flavored DDL (libSQL accepts it natively).

Migrations: libSQL adapter does NOT run the SQLite-specific ALTER TABLE migrations from migrateAlerts/migrateRegistryCheckConstraint because Turso deployments are GREENFIELD at v0.6.0 — fresh databases get the canonical schema directly. Future migration support lands in a dedicated phase if/when needed.

PragmaInit: no-op for libSQL — Turso server pre-configures journal_mode + busy_timeout. Multi-tenant Turso instances may reject PRAGMA writes from clients; skipping the round-trip avoids spurious errors.

Ping: synchronous round-trip via SELECT 1 to verify the URL is reachable + auth token valid. Wraps the underlying libsql error.

func OpenPostgresDB added in v0.4.0 

func OpenPostgresDB(url string) (*DB, error)

OpenPostgresDB opens (or connects to) the Postgres database at the given connection URL and ensures schema exists.

Phase 2.2 deliverable: makes Postgres a real backend option alongside the SQLite OpenDB. The returned *DB is fully interchangeable with the SQLite *DB at the SQLDB interface level (ExecDDL/ExecInsert/QueryRow/ RawQuery/Close/Ping/SetEncryptionKey). dialect.go helpers (TableExists/ColumnExists/PragmaInit/SchemaDDL) all dispatch correctly from the *DB.driver field set here.

URL format: standard libpq form supported by pgx, e.g.: 

postgres://user:pass@host:5432/dbname?sslmode=disable
postgres://user:pass@host/dbname (production with SSL)

Connection pool: pgx/v5/stdlib registers as the "pgx" sql.Driver; the returned *sql.DB uses Go's default pool settings (MaxOpenConns=0 meaning unlimited). Postgres handles concurrent connections natively (unlike SQLite's single-writer constraint), so we do NOT cap at 1 like OpenDB does — that would defeat the point of using Postgres.

Schema bootstrap: applies SchemaDDL(DialectPostgres) at open time. The Postgres-flavored DDL maps SQLite REAL → DOUBLE PRECISION, SQLite-INTEGER booleans (0/1) → BOOLEAN, and uses BIGINT-style identifiers in places where SQLite would use INTEGER PRIMARY KEY AUTOINCREMENT (none currently in alerts schema; relevant only for kite-mcp-audit's tool_calls/consent_log tables which use surrogate IDs).

Migrations: Postgres adapter does NOT run the SQLite-specific ALTER TABLE migrations from migrateAlerts/migrateRegistryCheckConstraint. Postgres deployments are GREENFIELD at v0.4.0 — fresh databases get the canonical schema directly. Future migration support (when existing Postgres deployments need schema evolution) lands in a dedicated phase, not v0.4.0.

Ping: synchronous round-trip to verify the URL is reachable. Returns the underlying pgx error wrapped with context.

func (*DB) Close 

func (d *DB) Close() error

Close closes the underlying database connection.

func (*DB) ColumnExists added in v0.3.1 

func (d *DB) ColumnExists(table, column string) (bool, error)

ColumnExists is the *alerts.DB-level wrapper for the package-level ColumnExists helper. Phase 2.2: dispatches via d.Dialect() — same pattern as TableExists.

func (*DB) DeactivateTrailingStop 

func (d *DB) DeactivateTrailingStop(id string) error

DeactivateTrailingStop marks a trailing stop as inactive.

func (*DB) DeleteAlert 

func (d *DB) DeleteAlert(email, alertID string) error

DeleteAlert removes an alert by ID for the given email.

func (*DB) DeleteAlertsByEmail 

func (d *DB) DeleteAlertsByEmail(email string) error

DeleteAlertsByEmail removes all alerts for the given email.

func (*DB) DeleteClient 

func (d *DB) DeleteClient(clientID string) error

DeleteClient removes an OAuth client by ID.

func (*DB) DeleteCredential 

func (d *DB) DeleteCredential(email string) error

DeleteCredential removes Kite credentials for the given email.

func (*DB) DeleteRegistryEntry 

func (d *DB) DeleteRegistryEntry(id string) error

DeleteRegistryEntry removes an app registration by ID.

func (*DB) DeleteSession 

func (d *DB) DeleteSession(sessionID string) error

DeleteSession removes an MCP session by ID. The session_id is hashed before lookup to match the HMAC-hashed PK in the database.

func (*DB) DeleteTelegramChatID 

func (d *DB) DeleteTelegramChatID(email string) error

DeleteTelegramChatID removes the Telegram chat ID mapping for the given email.

func (*DB) DeleteToken 

func (d *DB) DeleteToken(email string) error

DeleteToken removes a cached token for the given email.

func (*DB) Dialect added in v0.4.0 

func (d *DB) Dialect() Dialect

Dialect returns the dialect of the open connection. Used by dialect.go helpers (TableExists/ColumnExists/PragmaInit) to route catalog queries through the correct backend syntax. Phase 2.1.6 wrappers on *DB call this to pick their dispatch.

Always non-empty after OpenDB / OpenPostgresDB / OpenLibSQL returns successfully; nil receiver returns DialectSQLite as a backwards-compat default for pre-Phase-2.2 code paths that constructed *DB without going through the constructors.

func (*DB) ExecDDL 

func (d *DB) ExecDDL(ddl string) error

ExecDDL runs a DDL statement (CREATE TABLE, CREATE INDEX, etc.).

DDL statements never use `?` placeholders so no dialect rewrite is applied — the query goes through verbatim. Both dialects accept the same CREATE/ALTER/DROP syntax; dialect-specific schema choices are handled at the SchemaDDL helper layer (see dialect.go).

func (*DB) ExecInsert 

func (d *DB) ExecInsert(query string, args ...any) error

ExecInsert runs an INSERT (or similar DML) statement with arguments. Phase 2.4: rewrites `?` placeholders to `$N` form when the connection is Postgres. SQLite path: query passes through unchanged.

func (*DB) ExecResult 

func (d *DB) ExecResult(query string, args ...any) (sql.Result, error)

ExecResult runs a DML statement and returns the sql.Result for inspecting rows affected. Phase 2.4: dialect-aware placeholder rewrite.

func (*DB) GetConfig 

func (d *DB) GetConfig(key string) (string, error)

GetConfig retrieves a value from the config table by key. Returns ("", sql.ErrNoRows) if the key does not exist. Phase 2.4: routes through runQueryRow for dialect-aware placeholder rewrite.

func (*DB) LoadAlerts 

func (d *DB) LoadAlerts() (map[string][]*Alert, error)

LoadAlerts reads all alerts from the database grouped by email. Composite columns (alert_type, composite_logic, composite_name, conditions_json) are read via COALESCE so pre-composite rows decode cleanly with AlertTypeSingle. A malformed conditions_json fails loudly so data corruption is surfaced rather than silently producing a broken Alert — callers decide whether to halt startup or skip the row.

func (*DB) LoadClients 

func (d *DB) LoadClients() ([]*ClientDBEntry, error)

LoadClients reads all OAuth clients from the database. If an encryption key is set, client_secret is decrypted transparently. Pre-encryption plaintext values are returned as-is (migration-safe).

func (*DB) LoadCredentials 

func (d *DB) LoadCredentials() ([]*CredentialEntry, error)

LoadCredentials reads all stored Kite credentials from the database. If an encryption key is set, api_key and api_secret are decrypted transparently. Pre-encryption plaintext values are returned as-is (migration-safe).

func (*DB) LoadDailyPnL 

func (d *DB) LoadDailyPnL(email, fromDate, toDate string) ([]*DailyPnLEntry, error)

LoadDailyPnL reads daily P&L entries for a user within a date range (inclusive). Dates are in "2006-01-02" format.

Currency-aware (Slice 6d): scans the holdings/positions/net pnl currency labels alongside the float magnitudes. Existing rows pre-migration backfill to 'INR' via the ALTER TABLE DEFAULT (see db_migrations.go), so callers always observe a non-empty Currency string on a successful Load.

func (*DB) LoadRegistryEntries 

func (d *DB) LoadRegistryEntries() (map[string]*RegistryDBEntry, error)

LoadRegistryEntries reads all app registrations from the database. If an encryption key is set, api_secret is decrypted transparently.

func (*DB) LoadSessions 

func (d *DB) LoadSessions() ([]*SessionDBEntry, error)

LoadSessions reads all MCP sessions from the database. When encryption is enabled, session_id (PK) is an HMAC hash and the original session ID is recovered by decrypting session_id_enc. Rows without a valid encrypted ID are skipped (stale pre-migration data).

func (*DB) LoadTelegramChatIDs 

func (d *DB) LoadTelegramChatIDs() (map[string]int64, error)

LoadTelegramChatIDs reads all email-to-chatID mappings.

func (*DB) LoadTokens 

func (d *DB) LoadTokens() ([]*TokenEntry, error)

LoadTokens reads all cached Kite tokens from the database. If an encryption key is set, access tokens are decrypted transparently. Pre-encryption plaintext values are returned as-is (migration-safe).

func (*DB) LoadTrailingStops 

func (d *DB) LoadTrailingStops() ([]*TrailingStop, error)

LoadTrailingStops reads all active trailing stops from the database.

func (*DB) Ping 

func (d *DB) Ping() error

Ping verifies the SQLite connection is alive by issuing a no-op query. Used by /healthz?probe=deep to surface DB-side outages (file deleted, disk full, locked) that the in-process state alone cannot detect.

We use a simple SELECT 1 rather than db.Ping() because modernc.org/sqlite returns nil from Ping even on detached file handles in some scenarios; a real round-trip query catches more failure modes.

func (*DB) QueryRow 

func (d *DB) QueryRow(query string, args ...any) *sql.Row

QueryRow runs a query expected to return at most one row. Phase 2.4: dialect-aware placeholder rewrite.

func (*DB) RawQuery 

func (d *DB) RawQuery(query string, args ...any) (*sql.Rows, error)

RawQuery runs a query that returns rows. Phase 2.4: dialect-aware placeholder rewrite.

func (*DB) SaveAlert 

func (d *DB) SaveAlert(alert *Alert) error

SaveAlert inserts or replaces an alert in the database. Both single-leg and composite alerts route through here — the composite_* columns stay NULL for single-leg alerts, and the top-level Direction/TargetPrice are zero-valued (but present) for composite alerts so the NOT NULL columns are satisfied.

func (*DB) SaveClient 

func (d *DB) SaveClient(clientID, clientSecret, redirectURIsJSON, clientName string, createdAt time.Time, isKiteKey bool) error

SaveClient stores or updates an OAuth client in the database. If an encryption key is set and client_secret is non-empty, it is encrypted at rest.

func (*DB) SaveCredential 

func (d *DB) SaveCredential(email, apiKey, apiSecret, appID string, storedAt time.Time) error

SaveCredential stores or updates Kite credentials for the given email. If an encryption key is set, api_key and api_secret are encrypted at rest.

func (*DB) SaveDailyPnL 

func (d *DB) SaveDailyPnL(entry *DailyPnLEntry) error

SaveDailyPnL inserts or replaces a daily P&L entry.

Currency-aware (Slice 6d): writes the holdings/positions/net pnl currency labels alongside the float magnitudes. Empty Currency fields on the struct normalize to 'INR' via pnlCurrencyOrINR — the DB CHECK is NOT NULL so we must always supply a value, and INR is the production default (gokiteconnect emits INR prices by contract).

func (*DB) SaveRegistryEntry 

func (d *DB) SaveRegistryEntry(e *RegistryDBEntry) error

SaveRegistryEntry stores or updates an app registration. If an encryption key is set, api_key and api_secret are encrypted at rest.

func (*DB) SaveSession 

func (d *DB) SaveSession(sessionID, email string, createdAt, expiresAt time.Time, terminated bool) error

SaveSession stores or updates an MCP session in the database. The session_id PK is stored as HMAC-SHA256(key, sessionID) so the original session ID never appears in plaintext in the database. The original ID is stored encrypted in session_id_enc for recovery on restart.

func (*DB) SaveTelegramChatID 

func (d *DB) SaveTelegramChatID(email string, chatID int64) error

SaveTelegramChatID stores or updates a Telegram chat ID for the given email.

func (*DB) SaveToken 

func (d *DB) SaveToken(email, accessToken, userID, userName string, storedAt time.Time) error

SaveToken stores or updates a Kite token for the given email. If an encryption key is set, the access token is encrypted at rest.

func (*DB) SaveTrailingStop 

func (d *DB) SaveTrailingStop(ts *TrailingStop) error

SaveTrailingStop inserts or replaces a trailing stop in the database.

func (*DB) SetConfig 

func (d *DB) SetConfig(key, value string) error

SetConfig stores or updates a key-value pair in the config table.

SQL portability: ON CONFLICT (key) DO UPDATE SET value = excluded.value is the dialect-portable upsert form (SQLite ≥3.24, Postgres ≥9.5) per kite-mcp-server Phase 2.1 audit. Single SQL string for both dialects.

func (*DB) SetEncryptionKey 

func (d *DB) SetEncryptionKey(key []byte)

SetEncryptionKey sets the AES-256 key used to encrypt/decrypt credentials at rest. Derived from OAUTH_JWT_SECRET via HKDF. If not set, credentials are stored in plaintext.

func (*DB) TableExists added in v0.3.1 

func (d *DB) TableExists(name string) (bool, error)

TableExists is the *alerts.DB-level wrapper for the package-level TableExists helper. Phase 2.2: dispatches via d.Dialect() so the query routes to sqlite_master vs information_schema.tables based on which constructor opened this *DB (OpenDB → SQLite, OpenPostgresDB → Postgres).

func (*DB) UpdateAlertNotification 

func (d *DB) UpdateAlertNotification(alertID string, sentAt time.Time) error

UpdateAlertNotification records when a Telegram notification was sent for an alert.

func (*DB) UpdateTrailingStop 

func (d *DB) UpdateTrailingStop(id string, hwm, currentStop float64, modifyCount int) error

UpdateTrailingStop updates the high water mark, current stop, and modify count.

func (*DB) UpdateTriggered 

func (d *DB) UpdateTriggered(alertID string, price float64, at time.Time) error

UpdateTriggered marks an alert as triggered with the given price and time.

type DailyPnLEntry 

type DailyPnLEntry struct {
	Date                 string  `json:"date"`
	Email                string  `json:"email"`
	HoldingsPnL          float64 `json:"holdings_pnl"`
	HoldingsPnLCurrency  string  `json:"-"`
	PositionsPnL         float64 `json:"positions_pnl"`
	PositionsPnLCurrency string  `json:"-"`
	NetPnL               float64 `json:"net_pnl"`
	NetPnLCurrency       string  `json:"-"`
	HoldingsCount        int     `json:"holdings_count"`
	TradesCount          int     `json:"trades_count"`
}

DailyPnLEntry represents a single day's P&L snapshot.

Currency-aware (sibling-column design): the float64 PnL fields carry the magnitude; the matching *Currency string fields carry the unit label. The two are joined into a domain.Money via the *Money() accessors below so the in-memory pipeline (GetJournal aggregations, briefing.go consumers) gets currency-mismatch detection while the JSON wire format stays unchanged for kc/ops/api_alerts.go chart consumers and SQLite SUM()/AVG() aggregation continues to work natively.

Empty Currency string is the "INR default" sentinel — both the SQL boundary (DEFAULT 'INR' on the column) and the *Money() accessors normalize empty → "INR". gokiteconnect emits INR-priced floats by contract, so production rows always carry "INR".

json:"-" on the Currency fields keeps the wire shape identical to the pre-migration JSON ({"holdings_pnl": 1234.56, ...}).

func (*DailyPnLEntry) HoldingsPnLMoney 

func (e *DailyPnLEntry) HoldingsPnLMoney() domain.Money

HoldingsPnLMoney returns the holdings P&L as a domain.Money value. Empty currency defaults to INR.

func (*DailyPnLEntry) NetPnLMoney 

func (e *DailyPnLEntry) NetPnLMoney() domain.Money

NetPnLMoney returns the net P&L as a domain.Money value. Empty currency defaults to INR.

func (*DailyPnLEntry) PositionsPnLMoney 

func (e *DailyPnLEntry) PositionsPnLMoney() domain.Money

PositionsPnLMoney returns the positions P&L as a domain.Money value. Empty currency defaults to INR.

type Dialect added in v0.3.0 

type Dialect string

Dialect identifies the database backend dialect for SQL dispatch.

At v0.3.0 only DialectSQLite is wired through. DialectPostgres is the Phase 2.2 (OpenPostgresDB constructor) deliverable — at this commit, helpers either error on the Postgres branch or return SQLite-flavored output (SchemaDDL only) so the type signature stays stable for the eventual Postgres wire-up without a future API break. 

const (
	// DialectSQLite — modernc.org/sqlite (cgo-free) is the local
	// embedded driver. WAL journal + busy_timeout=5000ms +
	// foreign_keys=ON via DSN _pragma=foreign_keys(1).
	DialectSQLite Dialect = "sqlite"

	// DialectPostgres — github.com/jackc/pgx/v5 via database/sql
	// stdlib. Phase 2.2 deliverable. Postgres-flavored DDL
	// (REAL → DOUBLE PRECISION, INTEGER → BIGINT) routes via
	// SchemaDDL(DialectPostgres). Catalog queries route via
	// information_schema; placeholders rewritten `?` → `$N` per
	// rewritePlaceholders.
	DialectPostgres Dialect = "postgres"

	// DialectLibSQL — github.com/tursodatabase/libsql-client-go/libsql
	// via database/sql stdlib. Phase 2.6 Path 6 deliverable.
	// libSQL is an SQLite fork (Turso) that accepts SQLite-flavored
	// SQL natively — including `?` placeholders, INSERT OR REPLACE/
	// IGNORE, ON CONFLICT, sqlite_master/pragma_table_info. So
	// helpers route libSQL through the SAME paths as DialectSQLite
	// at the SQL level. The only difference is connection: libSQL
	// is hosted (libsql:// URL with ?authToken=...), no PRAGMA
	// init needed (Turso server pre-configures journal_mode + busy_timeout).
	DialectLibSQL Dialect = "libsql"
)

type Direction 

type Direction = domain.Direction

Direction is an alias for domain.Direction.

type Evaluator 

type Evaluator struct {
	// contains filtered or unexported fields
}

Evaluator checks incoming ticks against active alerts.

Wave D Phase 3 Package 4 (Logger sweep): logger is typed as the kc/logger.Logger port. NewEvaluator takes *slog.Logger for caller compatibility (kc/manager_init.go) and converts at the boundary via logport.NewSlog. Internal log calls use context.Background() — Evaluator is invoked from a long-lived ticker goroutine with no request ctx in scope.

func NewEvaluator 

func NewEvaluator(store *Store, logger *slog.Logger) *Evaluator

NewEvaluator creates a new alert evaluator.

func (*Evaluator) Evaluate 

func (e *Evaluator) Evaluate(email string, tick models.Tick)

Evaluate checks if a tick triggers any active alerts for the given instrument.

type KiteClientFactory 

type KiteClientFactory interface {
	NewClientWithToken(apiKey, accessToken string) zerodha.KiteSDK
}

KiteClientFactory creates Kite API clients (mirrors kc.KiteClientFactory for briefing use). Returns the hexagonal zerodha.KiteSDK port rather than the concrete *kiteconnect.Client so briefing + pnl services can be exercised off-HTTP with zerodha.MockKiteSDK.

type KiteOrderModifier 

type KiteOrderModifier interface {
	ModifyOrder(variety string, orderID string, params kiteconnect.OrderParams) (kiteconnect.OrderResponse, error)
}

KiteOrderModifier abstracts the Kite API call to modify an order. This enables testing without a live Kite connection.

type NotifyCallback 

type NotifyCallback func(alert *Alert, currentPrice float64)

NotifyCallback is invoked when an alert is triggered.

type PnLJournalResult 

type PnLJournalResult struct {
	Entries       []*DailyPnLEntry `json:"entries"`
	TotalDays     int              `json:"total_days"`
	CumulativePnL float64          `json:"cumulative_pnl"`
	BestDay       *DailyPnLEntry   `json:"best_day,omitempty"`
	WorstDay      *DailyPnLEntry   `json:"worst_day,omitempty"`
	WinDays       int              `json:"win_days"`
	LossDays      int              `json:"loss_days"`
	CurrentStreak int              `json:"current_streak"` // positive = winning, negative = losing
	BestStreak    int              `json:"best_streak"`
	WorstStreak   int              `json:"worst_streak"`
	AvgDailyPnL   float64          `json:"avg_daily_pnl"`
}

PnLJournalResult holds the result of a P&L journal query.

type PnLSnapshotService 

type PnLSnapshotService struct {
	// contains filtered or unexported fields
}

PnLSnapshotService takes daily P&L snapshots and provides journal queries.

Wave D Phase 3 Package 4 (Logger sweep): logger is typed as the kc/logger.Logger port. NewPnLSnapshotService takes *slog.Logger for caller compatibility (kc/manager_init.go) and converts at the boundary via logport.NewSlog. Internal log calls use context.Background() — TakeSnapshots is invoked from scheduler goroutines with no request ctx in scope.

func NewPnLSnapshotService 

func NewPnLSnapshotService(db *DB, tokens TokenChecker, creds CredentialGetter, logger *slog.Logger) *PnLSnapshotService

NewPnLSnapshotService creates a new P&L snapshot service. Returns nil if db is nil.

func (*PnLSnapshotService) GetJournal 

func (s *PnLSnapshotService) GetJournal(email, fromDate, toDate string) (*PnLJournalResult, error)

GetJournal retrieves P&L journal data for a user within a date range.

Currency-aware aggregation (Slice 6d): NetPnL summation routes through domain.Money.Add so cross-currency entries surface as a typed error rather than silently coercing INR + USD as a bare float. Production is INR-only by gokiteconnect contract — this is forward-compat guardrail for multi-currency Kite accounts.

Cumulative + AvgDailyPnL still surface as float64 on the result (JSON wire compat); the Money pipeline is the validation oracle, the scalar values come from .Float64() at the boundary.

func (*PnLSnapshotService) SetBrokerProvider 

func (s *PnLSnapshotService) SetBrokerProvider(p BrokerDataProvider)

SetBrokerProvider overrides the default Kite API client (for testing).

func (*PnLSnapshotService) SetKiteClientFactory 

func (s *PnLSnapshotService) SetKiteClientFactory(f KiteClientFactory)

SetKiteClientFactory wires the factory used by the default broker provider. Production wires this during app bootstrap; tests may leave it nil when they override the broker provider via SetBrokerProvider.

func (*PnLSnapshotService) TakeSnapshots 

func (s *PnLSnapshotService) TakeSnapshots()

TakeSnapshots captures daily P&L for all users with valid Kite tokens. Called by the scheduler at 3:40 PM IST.

type RegistryDBEntry 

type RegistryDBEntry struct {
	ID           string
	APIKey       string
	APISecret    string
	AssignedTo   string
	Label        string
	Status       string
	RegisteredBy string
	Source       string
	LastUsedAt   *time.Time
	CreatedAt    time.Time
	UpdatedAt    time.Time
}

RegistryDBEntry represents an app registration stored in the database.

type SQLDB 

type SQLDB interface {
	ExecDDL(ddl string) error
	ExecInsert(query string, args ...any) error
	ExecResult(query string, args ...any) (sql.Result, error)
	QueryRow(query string, args ...any) *sql.Row
	RawQuery(query string, args ...any) (*sql.Rows, error)
	Close() error
	Ping() error
	SetEncryptionKey(key []byte)
}

SQLDB is the dialect-portable surface that *DB exposes. Captures the driver-level operations callers consume (ExecDDL/ExecInsert/ExecResult/ QueryRow/RawQuery/Close/Ping/SetEncryptionKey) without coupling to dialect-specific helpers.

Phase 2.1 SQL portability work (kite-mcp-server commit da91a39, repo commits in v0.2.0 of each algo2go persistence repo): all production upsert sites converted from SQLite-only INSERT OR REPLACE / INSERT OR IGNORE to the portable ON CONFLICT (...) DO UPDATE / DO NOTHING form (SQLite ≥3.24, Postgres ≥9.5). GetConfig/SetConfig now use the portable ON CONFLICT form too — no longer SQLite-only.

Remaining SQLite-isms reserved for Phase 2.1.6 (dialect.go helper):

  • PRAGMA dispatch (WAL/busy_timeout/foreign_keys) — Postgres no-op
  • Schema DDL (REAL → DOUBLE PRECISION cosmetic, INTEGER PRIMARY KEY AUTOINCREMENT in audit → BIGSERIAL in Postgres)
  • sqlite_master query in db_migrations.go:15 (1 site)
  • billing self-migration INSERT OR IGNORE (column-resolution swallowing semantics not replicable in Postgres)

Postgres-readiness contract: a future Postgres adapter ships as a new struct (e.g. PostgresDB) implementing this SQLDB interface plus its own PRAGMA-dispatch and schema-introspection helpers. The driver-level method surface stays identical.

type SessionDBEntry 

type SessionDBEntry struct {
	SessionID  string
	Email      string
	CreatedAt  time.Time
	ExpiresAt  time.Time
	Terminated bool
}

SessionDBEntry represents an MCP session stored in the database.

type Store 

type Store struct {
	// contains filtered or unexported fields
}

Store is a thread-safe in-memory store for price alerts and Telegram chat IDs. Optionally backed by SQLite for persistence via SetDB.

Wave D Phase 3 Package 4 (Logger sweep): logger is typed as the kc/logger.Logger port. SetLogger keeps the *slog.Logger entry point for caller compatibility (kc/manager_init.go) and converts at the boundary via logport.NewSlog. SetLoggerPort is the port-typed alternative for callers already holding a logport.Logger. Internal log calls use context.Background() — Store methods are invoked from many call sites (HTTP handlers, ticker goroutines, scheduler) and the persistence-layer error logs don't carry a meaningful request ctx that's worth threading through every Add/Delete sig.

func NewStore 

func NewStore(onNotify NotifyCallback) *Store

NewStore creates a new alert store.

func (*Store) ActiveCount 

func (s *Store) ActiveCount(email string) int

ActiveCount returns the number of active (non-triggered) alerts for a user.

func (*Store) Add 

func (s *Store) Add(email, tradingsymbol, exchange string, instrumentToken uint32, targetPrice float64, direction Direction) (string, error)

Add creates a new alert and returns its ID. Returns an error if the user already has MaxAlertsPerUser alerts.

func (*Store) AddComposite 

func (s *Store) AddComposite(email, name string, logic CompositeLogic, conds []CompositeCondition) (string, error)

AddComposite creates a new composite alert — an alert that combines 2+ per-instrument conditions via AND/ANY logic — and returns its ID.

Composite alerts live in the same `alerts` table as single-leg alerts (Option B from the session handoff) with alert_type='composite', a JSON-encoded conditions payload, and the top-level Direction/TargetPrice fields intentionally zero — the evaluator walks Conditions instead.

Returns an error if conditions is empty or the user has reached MaxAlertsPerUser. Business-logic validation (min/max legs, operator compatibility, reference-price requirements) lives in the use case and is assumed to have already run — this method only guards persistence invariants.

func (*Store) AddWithReferencePrice 

func (s *Store) AddWithReferencePrice(email, tradingsymbol, exchange string, instrumentToken uint32, targetPrice float64, direction Direction, referencePrice float64) (string, error)

AddWithReferencePrice creates a new alert with an optional reference price (for percentage alerts) and returns its ID. Returns an error if the user already has MaxAlertsPerUser alerts.

func (*Store) Delete 

func (s *Store) Delete(email, alertID string) error

Delete removes an alert by ID for the given email.

func (*Store) DeleteByEmail 

func (s *Store) DeleteByEmail(email string)

DeleteByEmail removes all alerts for the given email. Used during account deletion to clean up all user data.

func (*Store) GetByToken 

func (s *Store) GetByToken(instrumentToken uint32) []*Alert

GetByToken returns all active (non-triggered) alerts matching the instrument token. Returns copies to prevent callers from mutating shared state.

func (*Store) GetEmailByChatID 

func (s *Store) GetEmailByChatID(chatID int64) (string, bool)

GetEmailByChatID performs a reverse lookup: given a Telegram chat ID, returns the associated email. Returns ("", false) if not found.

func (*Store) GetTelegramChatID 

func (s *Store) GetTelegramChatID(email string) (int64, bool)

GetTelegramChatID returns the Telegram chat ID for a user.

func (*Store) List 

func (s *Store) List(email string) []*Alert

List returns all alerts for the given email. Returns deep copies to prevent callers from mutating shared state.

func (*Store) ListAll 

func (s *Store) ListAll() map[string][]*Alert

ListAll returns a deep copy of all alerts grouped by email. Returns deep copies to prevent callers from mutating shared state.

func (*Store) ListAllTelegram 

func (s *Store) ListAllTelegram() map[string]int64

ListAllTelegram returns a copy of all Telegram chat ID mappings.

func (*Store) LoadFromDB 

func (s *Store) LoadFromDB() error

LoadFromDB populates the in-memory store from the database.

func (*Store) MarkNotificationSent 

func (s *Store) MarkNotificationSent(alertID string, sentAt time.Time)

MarkNotificationSent records when a Telegram notification was sent for an alert.

func (*Store) MarkTriggered 

func (s *Store) MarkTriggered(alertID string, currentPrice float64) bool

MarkTriggered marks an alert as triggered with the current price. Returns true if the alert was newly triggered, false if already triggered or not found.

func (*Store) SetDB 

func (s *Store) SetDB(db *DB)

SetDB enables write-through persistence to the given SQLite database.

func (*Store) SetLogger 

func (s *Store) SetLogger(logger *slog.Logger)

SetLogger sets the logger for DB error reporting.

Deprecated shim (Wave D Phase 3 Package 4): forwards to SetLoggerPort after wrapping via logport.NewSlog. Callers already on the port API should call SetLoggerPort directly.

func (*Store) SetLoggerPort 

func (s *Store) SetLoggerPort(logger logport.Logger)

SetLoggerPort sets the logger using the kc/logger.Logger port directly.

func (*Store) SetTelegramChatID 

func (s *Store) SetTelegramChatID(email string, chatID int64)

SetTelegramChatID sets the Telegram chat ID for a user.

type TelegramNotifier 

type TelegramNotifier struct {
	// contains filtered or unexported fields
}

TelegramNotifier sends alert notifications via Telegram.

Wave D Phase 3 Package 4 (Logger sweep): logger is typed as the kc/logger.Logger port. Constructors keep *slog.Logger parameters for caller compatibility (kc/manager_init.go) and convert at the boundary via logport.NewSlog. The Logger() accessor preserves its *slog.Logger return type via logport.AsSlog so existing tests (assert.Equal on the raw *slog.Logger) continue to pass — AsSlog returns the same instance NewSlog wrapped, not a copy.

func NewTelegramNotifier 

func NewTelegramNotifier(botToken string, store *Store, logger *slog.Logger) (*TelegramNotifier, error)

NewTelegramNotifier creates a new Telegram notifier using the package-level newBotFunc factory. Returns nil if botToken is empty (Telegram notifications disabled).

Prefer NewTelegramNotifierWithFactory when wiring tests that need a fake Telegram server — the per-call factory bypasses the global mutex on newBotFunc and is t.Parallel-safe.

func NewTelegramNotifierWithFactory 

func NewTelegramNotifierWithFactory(botToken string, store *Store, logger *slog.Logger, factory BotFactory) (*TelegramNotifier, error)

NewTelegramNotifierWithFactory creates a Telegram notifier using a caller-supplied BotFactory instead of the package-level newBotFunc global. Returns nil if botToken is empty (Telegram notifications disabled).

Tests inject a fake-server-backed factory here to keep the kc/alerts global state untouched — multiple parallel tests can each carry their own factory without racing on newBotFuncMu.

func (*TelegramNotifier) Bot 

func (t *TelegramNotifier) Bot() BotAPI

Bot returns the underlying Telegram bot API instance.

func (*TelegramNotifier) Logger 

func (t *TelegramNotifier) Logger() *slog.Logger

Logger returns the notifier's logger as a *slog.Logger.

Wave D Phase 3 Package 4 (Logger sweep): the underlying field is now a logport.Logger; we extract via logport.AsSlog. Returns nil if the notifier was constructed with a non-slogAdapter port (no production path does this — only mock-Logger tests, which don't call Logger()).

func (*TelegramNotifier) LoggerPort 

func (t *TelegramNotifier) LoggerPort() logport.Logger

LoggerPort returns the notifier's logger as a kc/logger.Logger port. New code should prefer this over Logger().

func (*TelegramNotifier) Notify 

func (t *TelegramNotifier) Notify(alert *Alert, currentPrice float64)

Notify sends a price alert notification to the user's Telegram.

func (*TelegramNotifier) SendHTMLMessage 

func (t *TelegramNotifier) SendHTMLMessage(chatID int64, text string) error

SendHTMLMessage sends an arbitrary HTML-formatted message to a specific chat.

func (*TelegramNotifier) SendMessage 

func (t *TelegramNotifier) SendMessage(chatID int64, text string) error

SendMessage sends an arbitrary MarkdownV2 message to a specific chat. Returns an error if the send fails. Callers are responsible for escaping user-supplied text with EscapeMarkdown before embedding it.

func (*TelegramNotifier) Store 

func (t *TelegramNotifier) Store() *Store

Store returns the underlying alert store (used by briefing to iterate users).

type TokenChecker 

type TokenChecker interface {
	// GetToken returns the access token and stored-at time for an email.
	// ok is false if no token exists.
	GetToken(email string) (accessToken string, storedAt time.Time, ok bool)
	// IsExpired returns true if a token stored at the given time has expired.
	IsExpired(storedAt time.Time) bool
}

TokenChecker abstracts the ability to look up a user's Kite token and check expiry.

type TokenEntry 

type TokenEntry struct {
	Email       string
	AccessToken string
	UserID      string
	UserName    string
	StoredAt    time.Time
}

TokenEntry represents a Kite access token stored in the database.

type TrailingStop 

type TrailingStop struct {
	ID              string    `json:"id"`
	Email           string    `json:"email"`
	Exchange        string    `json:"exchange"`
	Tradingsymbol   string    `json:"tradingsymbol"`
	InstrumentToken uint32    `json:"instrument_token"`
	OrderID         string    `json:"order_id"`        // the SL order to modify
	Variety         string    `json:"variety"`         // order variety (regular, co, etc.)
	TrailAmount     float64   `json:"trail_amount"`    // absolute trail in rupees (e.g., 20.0)
	TrailPct        float64   `json:"trail_pct"`       // OR percentage trail (e.g., 1.5 for 1.5%)
	Direction       string    `json:"direction"`       // "long" (trail up) or "short" (trail down)
	HighWaterMark   float64   `json:"high_water_mark"` // best price seen since activation
	CurrentStop     float64   `json:"current_stop"`    // current stop-loss trigger price
	Active          bool      `json:"active"`
	CreatedAt       time.Time `json:"created_at"`
	DeactivatedAt   time.Time `json:"deactivated_at,omitempty"`
	ModifyCount     int       `json:"modify_count"` // number of times the SL order was modified
	LastModifiedAt  time.Time `json:"last_modified_at,omitempty"`
}

TrailingStop represents a trailing stop-loss that modifies an existing SL order as the price moves favorably.

type TrailingStopManager 

type TrailingStopManager struct {
	// contains filtered or unexported fields
}

TrailingStopManager manages active trailing stops, evaluates ticks, and modifies SL orders via the Kite API when the trailing stop moves.

func NewTrailingStopManager 

func NewTrailingStopManager(logger *slog.Logger) *TrailingStopManager

NewTrailingStopManager creates a new trailing stop manager.

func (*TrailingStopManager) Add 

func (m *TrailingStopManager) Add(ts *TrailingStop) (string, error)

Add creates a new trailing stop and returns its ID.

func (*TrailingStopManager) Cancel 

func (m *TrailingStopManager) Cancel(email, id string) error

Cancel deactivates a trailing stop.

func (*TrailingStopManager) CancelByEmail 

func (m *TrailingStopManager) CancelByEmail(email string)

CancelByEmail deactivates all trailing stops for the given email. Used during account deletion to clean up all user data.

func (*TrailingStopManager) Evaluate 

func (m *TrailingStopManager) Evaluate(email string, tick models.Tick)

Evaluate processes a tick and updates trailing stops for the given instrument. This is called from the ticker's OnTick callback.

func (*TrailingStopManager) List 

func (m *TrailingStopManager) List(email string) []*TrailingStop

List returns all trailing stops for the given email.

func (*TrailingStopManager) LoadFromDB 

func (m *TrailingStopManager) LoadFromDB() error

LoadFromDB loads active trailing stops from the database.

func (*TrailingStopManager) SetDB 

func (m *TrailingStopManager) SetDB(db *DB)

SetDB enables write-through persistence.

func (*TrailingStopManager) SetEventDispatcher 

func (m *TrailingStopManager) SetEventDispatcher(d *domain.EventDispatcher)

SetEventDispatcher wires the typed domain event dispatcher so successful trailing-stop triggers emit TrailingStopTriggeredEvent. Nil-safe: when unset, only the slog Info log + onModify callback run (preserves the pre-ES behaviour for tests / bootstrap configurations that don't wire the dispatcher).

func (*TrailingStopManager) SetModifier 

func (m *TrailingStopManager) SetModifier(fn func(email string) (KiteOrderModifier, error))

SetModifier sets the function that provides a KiteOrderModifier for a given email.

func (*TrailingStopManager) SetOnModify 

func (m *TrailingStopManager) SetOnModify(fn func(ts *TrailingStop, oldStop, newStop float64))

SetOnModify sets the callback invoked after a successful SL order modification.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.