Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 14 additions & 4 deletions docs/strategy_plugin_runtime_contract.md
Original file line number Diff line number Diff line change
Expand Up @@ -124,13 +124,23 @@ when any of the following is true:
- `suggested_action` is `defend` or `blocked`
- `would_trade_if_enabled` is `true`

Platforms may still choose their delivery sinks, but Google Voice escalation over
SMTP should use
Platforms may still choose their delivery sinks, but Google Voice escalation via
the Gmail-to-Google-Voice SMS gateway should use
`quant_platform_kit.notifications.strategy_plugin_google_voice.publish_strategy_plugin_google_voice_alerts()`.
The publisher builds the shared subject/body, prefixes platform context, returns
structured sent/skipped/failed diagnostics, and can use
`StrategyPluginGoogleVoiceAlertMarkerStore` to skip alert keys that were already
sent. Platforms should expose this as Google Voice notification config, not as a
generic email alert surface.
sent.

Platforms should expose this as Google Voice notification config, not as a
generic email alert surface. The public configuration names should be channel
specific:

- `CRISIS_ALERT_GOOGLE_VOICE_GATEWAY`
- `CRISIS_ALERT_GOOGLE_VOICE_GMAIL_USER`
- `CRISIS_ALERT_GOOGLE_VOICE_GMAIL_APP_PASSWORD`

Future direct email notifications should use a separate namespace such as
`CRISIS_ALERT_EMAIL_*`.
This keeps the Crisis Response plugin behavior consistent across IBKR, Schwab,
LongBridge, Firstrade, and future platform runtimes.
Original file line number Diff line number Diff line change
Expand Up @@ -20,39 +20,35 @@

@dataclass(frozen=True)
class StrategyPluginGoogleVoiceSettings:
smtp_host: str | None = None
smtp_port: int = 587
sender: str | None = None
recipients: tuple[str, ...] = ()
username: str | None = None
password: str | None = field(default=None, repr=False)
use_starttls: bool = True
use_ssl: bool = False
gateway_recipients: tuple[str, ...] = ()
gmail_user: str | None = None
gmail_app_password: str | None = field(default=None, repr=False)
smtp_host: str = "smtp.gmail.com"
smtp_port: int = 465
use_starttls: bool = False
use_ssl: bool = True
timeout: float = 10.0

@classmethod
def from_object(cls, value: object) -> "StrategyPluginGoogleVoiceSettings":
if isinstance(value, cls):
return value
return cls(
smtp_host=_get_value(value, "crisis_alert_smtp_host"),
smtp_port=int(_get_value(value, "crisis_alert_smtp_port", 587) or 587),
sender=_first_non_empty(_get_value(value, "crisis_alert_smtp_from")),
recipients=tuple(parse_email_recipients(_get_value(value, "crisis_alert_google_voice_to", ()))),
username=_get_value(value, "crisis_alert_smtp_username"),
password=_get_value(value, "crisis_alert_smtp_password"),
use_starttls=_coerce_bool(_get_value(value, "crisis_alert_smtp_starttls", True), default=True),
use_ssl=_coerce_bool(_get_value(value, "crisis_alert_smtp_ssl", False), default=False),
gateway_recipients=tuple(
parse_email_recipients(_get_value(value, "crisis_alert_google_voice_gateway", ()))
),
gmail_user=_first_non_empty(_get_value(value, "crisis_alert_google_voice_gmail_user")),
gmail_app_password=_get_value(value, "crisis_alert_google_voice_gmail_app_password"),
)

def missing_fields(self) -> tuple[str, ...]:
missing: list[str] = []
if not str(self.smtp_host or "").strip():
missing.append("CRISIS_ALERT_SMTP_HOST")
if not str(self.sender or "").strip():
missing.append("CRISIS_ALERT_SMTP_FROM")
if not parse_email_recipients(self.recipients):
missing.append("CRISIS_ALERT_GOOGLE_VOICE_TO")
if not parse_email_recipients(self.gateway_recipients):
missing.append("CRISIS_ALERT_GOOGLE_VOICE_GATEWAY")
if not str(self.gmail_user or "").strip():
missing.append("CRISIS_ALERT_GOOGLE_VOICE_GMAIL_USER")
if not str(self.gmail_app_password or "").strip():
missing.append("CRISIS_ALERT_GOOGLE_VOICE_GMAIL_APP_PASSWORD")
return tuple(missing)

@property
Expand Down Expand Up @@ -282,10 +278,10 @@ def _send_message(
body=message.body,
smtp_host=settings.smtp_host,
smtp_port=settings.smtp_port,
sender=settings.sender,
recipients=settings.recipients,
username=settings.username,
password=settings.password,
sender=settings.gmail_user,
recipients=settings.gateway_recipients,
username=settings.gmail_user,
password=settings.gmail_app_password,
use_starttls=settings.use_starttls,
use_ssl=settings.use_ssl,
timeout=settings.timeout,
Expand Down Expand Up @@ -367,17 +363,6 @@ def _first_non_empty(*values: Any) -> str | None:
return None


def _coerce_bool(value: Any, *, default: bool) -> bool:
if value is None:
return default
if isinstance(value, bool):
return value
text = str(value).strip().lower()
if not text:
return default
return text in {"1", "true", "yes", "y", "on"}


def _fallback_alert_key(message: StrategyPluginAlertMessage) -> str:
return "strategy_plugin_google_voice_alert/" + _clean_relative_key(message.subject or "unknown")

Expand Down
42 changes: 28 additions & 14 deletions tests/test_google_voice_notifications.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,9 @@ def test_publish_strategy_plugin_google_voice_alerts_skips_missing_config():
assert result.sent_count == 0
assert result.skipped_count == 1
assert result.deliveries[0].reason == "missing_google_voice_config"
assert "CRISIS_ALERT_SMTP_HOST" in result.deliveries[0].error
assert "CRISIS_ALERT_GOOGLE_VOICE_GATEWAY" in result.deliveries[0].error
assert "CRISIS_ALERT_GOOGLE_VOICE_GMAIL_USER" in result.deliveries[0].error
assert "CRISIS_ALERT_GOOGLE_VOICE_GMAIL_APP_PASSWORD" in result.deliveries[0].error
assert observed == []


Expand All @@ -97,9 +99,9 @@ def test_publish_strategy_plugin_google_voice_alerts_sends_and_records_marker(tm
result = publish_strategy_plugin_google_voice_alerts(
[_alert_signal()],
google_voice_settings=StrategyPluginGoogleVoiceSettings(
smtp_host="smtp.example.com",
sender="bot@example.com",
recipients=("risk@example.com",),
gateway_recipients=("risk@example.com",),
gmail_user="bot@example.com",
gmail_app_password="app-password",
),
strategy_label="TQQQ",
context_label="ibkr / paper / tqqq",
Expand All @@ -112,15 +114,23 @@ def test_publish_strategy_plugin_google_voice_alerts_sends_and_records_marker(tm
assert result.failed_count == 0
assert result.deliveries[0].alert_key
assert "[ibkr / paper / tqqq]" in observed[0]["subject"]
assert observed[0]["smtp_host"] == "smtp.gmail.com"
assert observed[0]["smtp_port"] == 465
assert observed[0]["sender"] == "bot@example.com"
assert observed[0]["recipients"] == ("risk@example.com",)
assert observed[0]["username"] == "bot@example.com"
assert observed[0]["password"] == "app-password"
assert observed[0]["use_starttls"] is False
assert observed[0]["use_ssl"] is True
assert store.has_alert(result.deliveries[0].alert_key)


def test_publish_strategy_plugin_google_voice_alerts_skips_duplicate_marker(tmp_path):
store = StrategyPluginGoogleVoiceAlertMarkerStore(local_dir=tmp_path)
settings = StrategyPluginGoogleVoiceSettings(
smtp_host="smtp.example.com",
sender="bot@example.com",
recipients=("risk@example.com",),
gateway_recipients=("risk@example.com",),
gmail_user="bot@example.com",
gmail_app_password="app-password",
)
first = publish_strategy_plugin_google_voice_alerts(
[_alert_signal()],
Expand Down Expand Up @@ -148,16 +158,20 @@ def test_publish_strategy_plugin_google_voice_alerts_skips_duplicate_marker(tmp_
assert second.deliveries[0].reason == "duplicate_alert"


def test_google_voice_settings_read_google_voice_names_only():
def test_google_voice_settings_reads_google_voice_gmail_names_only():
settings = StrategyPluginGoogleVoiceSettings.from_object(
SimpleNamespace(
crisis_alert_smtp_host="smtp.gmail.com",
crisis_alert_smtp_from="sender@gmail.com",
crisis_alert_google_voice_to="gateway@txt.voice.google.com",
crisis_alert_smtp_username="sender@gmail.com",
crisis_alert_google_voice_gateway="gateway@txt.voice.google.com",
crisis_alert_google_voice_gmail_user="sender@gmail.com",
crisis_alert_google_voice_gmail_app_password="app-password",
)
)

assert settings.sender == "sender@gmail.com"
assert settings.recipients == ("gateway@txt.voice.google.com",)
assert settings.gmail_user == "sender@gmail.com"
assert settings.gateway_recipients == ("gateway@txt.voice.google.com",)
assert settings.gmail_app_password == "app-password"
assert settings.smtp_host == "smtp.gmail.com"
assert settings.smtp_port == 465
assert settings.use_ssl is True
assert settings.use_starttls is False
assert settings.missing_fields() == ()