<强 > 更新2019年
实现这一点有几种办法,这里是按日益复杂程度排列的清单。
- Python Shell, you will write streams to the python console and it
will write back to you
- Redis Pub Sub, you can have a channel
listening in Python while your node js publisher pushes data
- Websocket connection where Node acts as the client and Python acts
as the server or vice-versa
- API connection with Express/Flask/Tornado etc working separately with an API endpoint exposed for the other to query
<强势>Approach 1 Python Shell 简单方法
< 强力> 源代码.js 文件
const ps = require( python-shell )
// very important to add -u option since our python script runs infinitely
var options = {
pythonPath: /Users/zup/.local/share/virtualenvs/python_shell_test-TJN5lQez/bin/python ,
pythonOptions: [ -u ], // get print results in real-time
// make sure you use an absolute path for scriptPath
scriptPath: "./subscriber/",
// args: [ value1 , value2 , value3 ],
mode: json
};
const shell = new ps.PythonShell("destination.py", options);
function generateArray() {
const list = []
for (let i = 0; i < 1000; i++) {
list.push(Math.random() * 1000)
}
return list
}
setInterval(() => {
shell.send(generateArray())
}, 1000);
shell.on("message", message => {
console.log(message);
})
< 强度 > 目的地 。 py 文件
import datetime
import sys
import time
import numpy
import talib
import timeit
import json
import logging
logging.basicConfig(format= %(asctime)s : %(levelname)s : %(message)s , level=logging.INFO)
size = 1000
p = 100
o = numpy.random.random(size)
h = numpy.random.random(size)
l = numpy.random.random(size)
c = numpy.random.random(size)
v = numpy.random.random(size)
def get_indicators(values):
# Return the RSI of the values sent from node.js
numpy_values = numpy.array(values, dtype=numpy.double)
return talib.func.RSI(numpy_values, 14)
for line in sys.stdin:
l = json.loads(line)
print(get_indicators(l))
# Without this step the output may not be immediately available in node
sys.stdout.flush()
< strong > notes : 制作一个名为用户的文件夹, 该文件夹与源代码. js 文件处于同一级别, 并将目标. py 放在其中 。 不要忘记改变您的虚拟环境 。
